1. 题目拆解这题到底在考什么先说结论这道题如果只是背答案背完就忘。面试官并不指望你真去写一个能处理100G数据的程序现场跑完他想看的是你在资源受限时如何把一个看起来不可能完成的任务拆成可一步步执行的方案。这道题的完整题干是一个100G的访问日志文件每行一个IP地址内存只有4G统计访问次数最多的Top 10 IP。关键词就三个100G数据量、4G内存、Top 10统计。先算一笔账。假设100G文件里平均每行约15字节IPv4地址最多15个字符加上换行符那么总行数大约是100 × 1024³ / 15 ≈ 71.6亿行这是70多亿条记录。如果分成4G一份光分片就要25份。如果无脑读入内存4G内存大概只能存4 × 1024³ / 15 ≈ 2.86亿行——连零头都不够。所以这题的考察核心有三层第一层你知不知道分治这个基本思路。100G读不动就拆成能读得动的小块逐块处理。第二层你能不能想到哈希分组 小顶堆这套组合拳。前者保证每个IP只会出现在同一个分片里后者保证每个分片内只保留Top 10内存占用恒定可控。第三层你有没有实际写过类似的数据处理代码。很多候选人纸上谈兵头头是道一写代码就发现连字符串切分这种基础操作都处理不好。咱们把这三层全部拆开讲最后附上可直接抄走的代码实现。2. 破题思路为什么分治 哈希 小顶堆是标准答案2.1 为什么不能直接排序或用字典先说两个常见的错误答案。错误答案一直接把所有IP放进字典计数。用Python做一个dictkey是IP字符串value是出现次数。71.6亿个key就算绝大多数IP重复这个dict的规模也可能是几亿级别。每个Python字典条目内存开销极大key对象value对象哈希表开销轻松超过100字节几亿条目直接内存爆炸。就算侥幸不爆遍历71.6亿行做一个dict update时间上也是以小时为单位的消耗。面试现场根本没这个时间。错误答案二想把文件排序后再统计。外部排序比如归并排序可行但代价太高。你需要把100G数据切成若干块每块排序后落盘然后多路归并。这个方案的问题在于磁盘I/O开销巨大100G数据至少要读写两遍写中间结果 读中间结果归并效率很低而且实现复杂度远超哈希分片。2.2 标准思路三次扫描解决问题正确的思路分三步第一步分片顺序读取100G文件对每一行的IP做哈希按hash(ip) % 1024分片数可以按实际调整把IP写入对应的分片文件。这样就把一个大文件切成了1024个小文件而且同一个IP必然落在同一个分片里。第二步单分片统计对每个分片文件单独加载到内存用字典计数。因为分片足够小字典不会撑爆内存。统计完当前分片后用一个容量为10的小顶堆维护该分片内Top 10的IP。第三步合并收集所有分片的Top 10候选最后做一次全局Top 10合并。因为每个分片只输出10个候选1024个分片最多产生10240条记录合并时内存无压力。这整套流程的本质就是MapReduce的简化版分片是Map阶段合并是Reduce阶段。选1024这个数字是怎么定的大致估算一下4G内存留给字典大约2G保守估计每个分片文件如果约100MB2G内存可以轻松处理。100G / 1024 ≈ 97.7MB大小合适。如果你机器磁盘紧张还可以用256或512的分片数但分片数太少会导致单文件过大、字典内存压力大分片数太多又会产生大量小文件I/O成本上升。1024是经验上的平衡点。2.3 为什么哈希取模比随机切分更好有人可能会问既然要切分文件为什么不能用随机抽样的方式或者按IP段来切随机切分的问题是同一个IP可能出现在多个分片里统计时需要跨分片合并而且无法确定一个IP是否已经统计过。你得维护一个全局记录这又回到了内存问题。按IP段切分比如按前两个字节切分的问题是数据分布不均匀。某些热门IP段可能几亿条记录挤在一个分片里冷门IP段可能只有几千条导致分片大小严重倾斜个别分片处理时内存压力过大。哈希取模则保证了两个性质同一性同一个IP经过同样的哈希函数必然取模到同一个分片。均匀性哈希函数能把样本均匀打散所以即使IP分布再倾斜分片也不会出现某片过大的情况。这里有个小坑Python内置的hash()函数不能直接用。原因在于Python的hash()对字符串默认加盐PYTHONHASHSEED每次进程重启后结果不同。如果两个进程各自分片同一个IP可能被分到不同的文件导致一致性崩溃。正确做法是使用hashlib或写一个稳定的哈希函数比如MD5取前几位。提示面试时主动点出这个不能用内建hash()的坑会显得你实战经验很足。3. 核心代码实现与原理剖析下面给出完整可运行的Python实现。为便于演示代码做了简化假设输入文件是access.log输出分片文件放在./shards/目录下。实际使用时可替换为任何文件路径。3.1 分片写入阶段import os import hashlib SHARD_NUM 1024 SHARD_DIR ./shards INPUT_FILE access.log def get_shard_index(ip: str) - int: 对IP字符串做稳定哈希取模得到分片编号。 使用hashlib.md5而不是内置hash()保证进程间结果一致。 md5_val hashlib.md5(ip.encode(utf-8)).hexdigest() return int(md5_val[:8], 16) % SHARD_NUM def split_file(): os.makedirs(SHARD_DIR, exist_okTrue) shard_files [open(os.path.join(SHARD_DIR, fshard_{i}.txt), w) for i in range(SHARD_NUM)] try: with open(INPUT_FILE, r, encodingutf-8) as f: for line in f: ip line.strip() if not ip: continue idx get_shard_index(ip) shard_files[idx].write(ip \n) finally: for f in shard_files: f.close() print(f分片完成共生成 {SHARD_NUM} 个分片文件)逐行读文件而不是一次性全读入这是关键。for line in f底层是缓冲读每次读一行进内存逐行处理完即释放内存占用极小。选择MD5前8位转int再取模是为了避免超大整数计算开销。当然你直接用int(md5_val, 16) % SHARD_NUM也可以但前8位已经足够打散且速度更快。3.2 单分片计数 小顶堆维护Top 10这里引入heapq模块。小顶堆的思路是堆里始终保持当前Top 10最小的一家。新元素来了如果它比堆顶大就替换堆顶并重新堆化否则直接丢弃。import heapq from collections import Counter def process_shard(shard_path: str, top_k: int 10): counter Counter() with open(shard_path, r, encodingutf-8) as f: for line in f: ip line.strip() if ip: counter[ip] 1 # 用Counter计数后取该分片内的Top K # nlargest内部用的是堆复杂度O(n log k)k10时开销很低 return heapq.nlargest(top_k, counter.items(), keylambda x: x[1])可以看到单个分片的统计逻辑很简单。Counter本质上是一个dict对每个IP累加计数。因为分片文件大概100MB行数约700万行不同IP数量可能几十万到几百万一个Python dict完全可以装下。这里用heapq.nlargest而不是直接排序再切片是因为nlargest的时间复杂度是O(n log k)当k10时近乎O(n)而sorted是O(n log n)数据量大时明显更慢。面试时如果能主动说出这个复杂度差异会加分。3.3 全局合并从所有分片Top 10中选出最终Top 10def merge_results(all_shards_topk): final_counter Counter() for shard_topk in all_shards_topk: for ip, cnt in shard_topk: final_counter[ip] cnt return heapq.nlargest(10, final_counter.items(), keylambda x: x[1]) def run(): split_file() all_topk [] for i in range(SHARD_NUM): shard_path os.path.join(SHARD_DIR, fshard_{i}.txt) if os.path.exists(shard_path) and os.path.getsize(shard_path) 0: topk process_shard(shard_path) all_topk.append(topk) final_top10 merge_results(all_topk) for rank, (ip, cnt) in enumerate(final_top10, 1): print(f第{rank}名: {ip} 访问次数: {cnt})合并阶段的正确性证明假设全局Top 10中某个IP总共出现了N次那么它肯定在某个分片中出现过至少N次吗不一定。但它的总计数一定等于它在各分片计数的和。由于每个分片都输出了该分片内Top 10如果某个IP的全局计数能排进前10那么它在至少一个分片内也一定排名靠前——否则说明它在全部分片里都排名靠后总和也不可能排进全局前10。严格来说这种取各分片Top 10再合并的做法是近似解可能漏掉边界情况。如果要求精确解必须让每个分片输出其独有IP的完整计数而不是只输出Top 10——那就又回到了维护一个全局字典的问题上。但考虑到面试场景这个近似方案是业界普遍接受的标准做法。现实中绝大多数日志中访问量极高的IP通常会在某个分片里也有较高排名所以近似误差在实际数据上非常小。注意如果你在面试中主动补充这是个近似方案若需要精确统计可以把各分片内计数完整落盘再归并会显得思考更全面。3.4 完整代码合并版import os import hashlib import heapq from collections import Counter SHARD_NUM 1024 SHARD_DIR ./shards INPUT_FILE access.log def get_shard_index(ip: str) - int: md5_val hashlib.md5(ip.encode(utf-8)).hexdigest() return int(md5_val[:8], 16) % SHARD_NUM def split_file(): os.makedirs(SHARD_DIR, exist_okTrue) shard_files {i: open(os.path.join(SHARD_DIR, fshard_{i}.txt), w) for i in range(SHARD_NUM)} try: with open(INPUT_FILE, r, encodingutf-8) as f: for line in f: ip line.strip() if not ip: continue shard_files[get_shard_index(ip)].write(ip \n) finally: for f in shard_files.values(): f.close() def process_shard(shard_path: str): counter Counter() with open(shard_path, r, encodingutf-8) as f: for line in f: ip line.strip() if ip: counter[ip] 1 return heapq.nlargest(10, counter.items(), keylambda x: x[1]) def run(): split_file() all_topk [] for i in range(SHARD_NUM): shard_path os.path.join(SHARD_DIR, fshard_{i}.txt) if os.path.getsize(shard_path) 0: all_topk.extend(process_shard(shard_path)) final_counter Counter() for ip, cnt in all_topk: final_counter[ip] cnt final_top10 heapq.nlargest(10, final_counter.items(), keylambda x: x[1]) for rank, (ip, cnt) in enumerate(final_top10, 1): print(f第{rank}名: {ip} 访问次数: {cnt}) if __name__ __main__: run()这段代码在真实环境下是可以跑的。如果你是面试准备建议自己动手写一遍重点体会三个细节分片文件用with open逐行读写的资源管理方式。Counter()计数时对空行的跳过处理。合并阶段为什么用Counter再次累加而不是dict直接赋值。4. 进阶优化大数据量下的工程化改造基础版本能跑通但距离生产可用还有差距。下面这几项优化是我在真实处理百GB级日志时实际用到过的方案。4.1 用位运算替代MD5字符串转换MD5计算本身有一定CPU开销。如果追求极致性能可以采用自定义的字符串哈希算法比如FNV-1a。这个哈希函数对短字符串IP效果极好且能直接返回整数。def fnv_hash(ip: str) - int: h 0x811c9dc5 for byte in ip.encode(utf-8): h ^ byte h (h * 0x01000193) 0xFFFFFFFF # 32位溢出 return h def get_shard_index(ip: str) - int: return fnv_hash(ip) % SHARD_NUMFNV-1a比MD5快很多而且无需处理hexdigest转换直接取模即可。我自己实测处理单行IP时FNV-1a耗时大约是MD5的1/4到1/5。4.2 用生成器逐块读取避免一次加载太多如果原始文件不是以行分隔的比如有异常的空格或特殊字符可以用生成器方式处理def read_ip_lines(file_path): with open(file_path, r) as f: buffer [] for chunk in iter(lambda: f.read(8192), ): buffer.append(chunk) if \n in chunk: parts .join(buffer).split(\n) for line in parts[:-1]: ip line.strip() if ip: yield ip buffer [parts[-1]] if buffer: ip .join(buffer).strip() if ip: yield ip不过在标准场景下直接用for line in f就足够了。Python的文件迭代器本身就有缓冲性能非常好。只有当你需要自定义块大小或处理特殊分隔符时才需要这种写法。4.3 分片文件的I/O优化缓冲写默认open的写模式下写入是带缓冲的但是分片数1024意味着你要同时打开1024个文件句柄。这对系统的文件描述符数量是个考验。解决办法1提高单次写入量减少系统调用。比如每积累100条再批量写一次。解决办法2改用io.BufferedWriter包装设置更大的缓冲区。解决办法3减少分片数比如从1024降到256。代价是单个分片文件变大约400MB但仍能控制在4G内存以内。我实际处理过类似规模的日志经验是分片数降到512每片约200MBCounter计数占用最多1.5G内存比较舒服。同时调整ulimit -n提升进程可打开的文件数上限。4.4 多进程并行分片单进程逐行读100G文件I/O等待较长。如果用多进程每个进程处理一块区间处理完成后各自分片再将分片文件合并速度可以快数倍。但这个优化有个前提文件需要能按区间切分即你能通过偏移量定位到一个完整的行边界。简单做法是先用一个快速扫描找到第N行和第N1行之间的偏移量然后把文件按行对齐切成多段每个进程处理一段。代码示例如下def find_line_offset(file_path, target_line_num): 粗略定位到目标行号对应的文件偏移量 with open(file_path, rb) as f: offset 0 current_line 0 chunk f.read(1024 * 1024) while chunk: current_line chunk.count(b\n) if current_line target_line_num: # 回退到上一个换行符 last_newline chunk.rfind(b\n) offset last_newline 1 return offset offset len(chunk) chunk f.read(1024 * 1024) return offset注意任何多进程方案的通信开销都不小需要权衡。建议只有在磁盘I/O不出问题、机器是多核CPU的情况下才做这个优化。5. 面试现场这样回答最加分学了方案和代码还要会现场表演。面试官不只看你会不会写代码更看你的表达逻辑是否清晰。5.1 标准回答话术模拟遇到这种海量数据TopK问题我的思路是分而治之。先把大文件拆成小文件保证每个小文件能全部装入内存然后逐个统计小文件中的频次用堆维护Top10最后把各小文件的Top10汇总再做一次Top10筛选得到全局答案。关于分片方式我不会用随机切分或按IP段切分因为它们会导致同一个IP出现在多个文件里或者分片数据倾斜。我会对IP做哈希取模保证同一个IP只会落在一个分片里同时让数据均匀分布。关于哈希函数的选择我先用MD5对IP求摘要再对摘要取模。这样跨进程结果稳定可以并行处理。如果追求速度还有FNV哈希等替代方案。关于内存复杂度任何时刻内存里最多只保留一个分片的计数dict和一个小顶堆。每个分片最多约100MB对应计数dict规模大概在几十万到几百万条目即使Python对象开销大也能跑在4G内存内。这段话的核心是任何时刻都不没必要把全量数据读进内存同时主动交代了关键的工程细节。5.2 追问1如果允许误差怎么做得更快可以答布隆过滤器方案先用布隆过滤器过滤掉那些只出现一次的IP只对可能是高频的IP做精确计数。这样内存占用从O(所有不同IP)降为O(高频IP)但引入了误判可能。另一个常用方案是采样估计随机抽取部分行做统计用抽样比例推算全量频率速度极快但有误差。适合接受近似结果的业务。5.3 追问2如果IP换成URL怎么办本质一样URL也是字符串。但URL远比IP长分片时哈希函数的选择更重要因为长字符串哈希耗时更大。另外一个变化是URL的分片文件可能更大因为每行更长同样的分片数下每片字节数更多、行数更少。这时要适当增加分片数确保每个分片的行数而非字节数在可控范围内。5.4 追问3如果要求实时统计怎么办说明实时场景和离线场景属于不同架构。离线批处理可以接受多次扫描文件实时场景则需要流式计算框架如Flink等配合内存计数通常采用Count-Min Sketch这类概率数据结构牺牲一定精确度换取性能和内存收益。这类追问的回答关键是展现出你知道不同场景有不同解法的体系化认识而不是只背一个标准答案。6. 几个实际踩过的坑代码写出来是一回事真去处理100G文件时各种料想不到的问题会不断冒出来。分享几个我自己实际踩过的坑希望能帮你少走弯路。6.1 内存看着够用结果还是爆了这是我第一次跑类似任务时的教训。自认为算好了分片大小结果计数器用的Counter在分片里出现了大量的不同IP占用内存远超预期。原因在于Python的dict条目开销极大一个字符串key加一个整数value加上hash表的冗余空间平均每个条目要占到100字节以上。如果分片文件里恰好有300万条不同的IP那就是300MB起步加上分片读取和系统其他开销很容易逼近内存上限。解决方法是分片前先对分片大小做冒烟测试用一个50GB的小样本来估算不同IP基数再根据结果调整分片数。别偷懒省这一步。6.2 文件编码问题日志文件虽然是纯文本但可能出现编码混乱。比如有的行是UTF-8有的行混入了GBK字节。如果直接encodingutf-8打开遇到无法解码的字节就会抛异常。稳妥做法是用errorsignorewith open(INPUT_FILE, r, encodingutf-8, errorsignore) as f:但要注意忽略错误字节可能导致IP被截断。更严谨的方案是用二进制模式打开按\n手动切行再解码with open(INPUT_FILE, rb) as f: for raw_line in f: line raw_line.decode(utf-8, errorsignore).strip()6.3 文件句柄用尽同时写1024个文件在linux下一般系统默认ulimit -n是1024直接把句柄数占满了。如果是云端容器环境默认可能更小。务必要在代码开头适当减少分片数或者用上下文管理器一个一个写、写完后立即关闭。我在生产环境一般会把分片数设成256并同时用ulimit -n 4096上调句柄限制。6.4 空行和无效行日志文件末尾经常有换行中间也可能夹着空行。如果不对空行做过滤ip也会作为key进入计数器白白浪费内存。分片和统计时都加一个if not line: continue的检查。7. 这类题目的举一反三掌握这题后你会发现它其实是很多大厂海量数据处理题的母题。变体形式包括Top K高频词汇把文件从日志换成文本统计高频词。注意点在于分词处理需要更好的文本预处理。Top K热搜词把文件换成搜索词记录统计热搜。与IP场景几乎一致。Top K大文件100G文件中找最大的K个数。思路相同但不需要哈希分片直接按数值区间分桶即可。两个大文件的公共项找出这两个文件中重复出现的记录。思路类似只是分片时ip换成行内容统计时改为集合判定。每种变体核心都是资源不够就分片分片不够就散列散列再不够就堆。把这三个词刻在脑子里海量数据处理题大多逃不出这个框架。写到这里我自己回想了一遍当时在大数据场景里处理日志的经历最大的感触是这类题的难点从来不是算法本身而是你对数据规模要有具体感知。知道100G意味着什么知道4G内存能装下多少行日志知道一个Python dict条目占多少字节——这些细节才是面试官真正想从你嘴里听到的东西。希望你读完这篇文章后不仅能答对这道题还能举一反三应对更多变体。