前阵子有个同事接了一个统计需求单台网关一天的访问日志有好几个GB需要在秒级算出访问量最高的100个URL。他第一版代码直接用readlines()把所有数据灌进内存然后全量排序结果内存告警机器差点卡死。后来我改成手写 TopK 算法内存占用从几个GB降到了几十MB问题迎刃而解。这个场景几乎每天都在发生数据量大于内存但我们只关心最大的K个值。这类题也常出现在各种算法考核里很多人能背出“用堆”但真到了工程环境面临数据量大于内存的前提时怎么写、怎么验证、有哪些隐藏的坑能把每一步都讲清楚的人其实不多。这篇就把完整链路拆开讲一遍。1. 先确认题目边界到底是大数据加小K还是 K 本身也放不下很多人一听到“数据量大于内存”第一反应就是上分布式计算或者上外部存储。实际上绝大多数 TopK 场景用单机堆就能解决关键是要先搞清楚“哪个东西放不下”。场景总数据量 NK 的大小内存约束推荐路线最常见的 TopK远大于内存远小于内存内存能放下 K 个元素和少量缓冲单机最小堆O(N log K)K 也很大远大于内存接近或大于内存连 K 个候选都放不下分块局部 TopK 合并 / 外部排序流式无界数据无线远小于内存只能顺序读一次在线堆值域有限的计数很大任意能放得下值域频次数组计数或桶排序O(N)这里的核心是我们平时说的“数据量大于内存”很多情况下是“N 远大于内存但 K 远小于内存”。比如从 10 亿条记录里取前 100堆只需要维护 100 个元素哪怕每条记录本身很大也完全放得下。真正的困难反而是 K 大到一定程度以后连堆都装不下这时才需要考虑外部归并和分块策略。1.1 为什么不能一上来就全排序全排序是一个非常直接的思路把数据全部读进内存排序然后取前 K 个。但问题有两个第一内存放不下。当数据量大于内存时全排序的readlines()或全量加载会直接 OOM就算用外部排序也要把整个 N 都写一遍磁盘再读一遍磁盘IO 成本高得离谱。第二做了大量无用功。全排序的时间复杂度是 O(N log N)它把 N 个元素的所有顺序关系都排清楚了但我们只需要前 K 个倒数的那些元素根本不需要关心。只要 K 远小于 N堆方案 O(N log K) 的时间复杂度优势就非常明显。1.2 一个容易被忽略的变体数据是流式的如果数据本身是无限流比如实时埋点、实时监控指标此时根本无法一次性拿到完整数据集更不可能先收集再排序。堆是天然的在线算法数据逐个进入堆内永远保存“当前已经见过的数据中最大的 K 个”新数据来了直接处理旧数据不需要保留。这也是为什么堆在流式计算框架里被广泛使用。所以做题或者做工程时第一步永远不是写代码而是确认数据边界N 有多大K 有多大是离线文件还是在线流这几个问题直接决定你该用哪套方案。2. 最小堆解法手写实现不依赖任何第三方库手写 TopK 算法核心是手写堆。堆本质上就是一棵完全二叉树用数组就能存。我先把最小堆实现出来再讲主流程。2.1 为什么求最大的 K 个要用最小堆这个点经常有人搞反。求最大的 K 个堆里存的是“当前已经见过的最大的 K 个元素”。在这 K 个元素里谁最弱是堆顶也就是这 K 个里最小的那个。新元素来了如果它比堆顶还小说明它连最弱的都打不过直接淘汰如果它比堆顶大就把它替换进堆让原来的堆顶成为历史。打个比方一个 K 人的精英名单名单最后一名是守门员。新人想进名单只要比守门员强就把守门员挤走如果连守门员都比不过那就没资格进来。求最小的 K 个则反过来用最大堆堆顶是当前 K 个里最大的那个新人更小才能挤进去。2.2 手写一个最小堆数组、上浮、下沉下面是完整的最小堆实现包括建堆、上浮、下沉和替换堆顶。这里刻意不用 Python 标准库目的是把堆的结构讲清楚。class MinHeap: def __init__(self, valuesNone): self.heap [] if values is not None: self.heap list(values) n len(self.heap) for i in range(n // 2 - 1, -1, -1): self._sift_down(i) def __len__(self): return len(self.heap) def peek(self): return self.heap[0] if self.heap else None def push(self, val): self.heap.append(val) self._sift_up(len(self.heap) - 1) def heapreplace(self, val): # 用 val 替换堆顶并调整调用方需要保证 val 当前堆顶 if self.heap: self.heap[0] val self._sift_down(0) def _sift_up(self, idx): while idx 0: parent (idx - 1) // 2 if self.heap[idx] self.heap[parent]: self.heap[idx], self.heap[parent] self.heap[parent], self.heap[idx] idx parent else: break def _sift_down(self, idx): n len(self.heap) while True: left 2 * idx 1 right 2 * idx 2 smallest idx if left n and self.heap[left] self.heap[smallest]: smallest left if right n and self.heap[right] self.heap[smallest]: smallest right if smallest idx: break self.heap[idx], self.heap[smallest] self.heap[smallest], self.heap[idx] idx smallestsift_up用于插入元素时把新元素从尾部不断和父节点比较如果比父节点小就交换保证堆性质。sift_down用于替换堆顶或者建堆时从某个位置开始往下调整找左右孩子中最小的那个如果比自己小就交换。这两个函数是所有堆操作的根基。2.3 主流程一次遍历两次判断有了最小堆之后TopK 主流程非常简单def topk_largest(iterable, k): if k 0: return [] it iter(iterable) heap [] # 先取前 K 个元素不足 K 个直接返回有序结果 try: for _ in range(k): heap.append(next(it)) except StopIteration: return sorted(heap, reverseTrue) # 批量建堆比逐个 push 更快 heap_min MinHeap(heap) for x in it: if x heap_min.peek(): heap_min.heapreplace(x) return sorted(heap_min.heap, reverseTrue)核心逻辑就一个遍历每个元素先跟堆顶比较只有大于堆顶才替换。整个过程最多只有一次建堆的 O(K)之后每个元素要么花 O(1) 做比较要么花 O(log K) 做下沉。注意heapreplace之前必须做比较否则会出大问题。2.4 比较与替换的次序一个容易写错的细节很多人写到这里会图省事省略if x heap_min.peek()直接调用heapreplace这是错的。我给你一个最直观的返例K1数据依次是 [5, 4, 3, 2, 1]。如果无条件替换初始堆是 [5]遇到 4 替换成 [4]遇到 3 替换成 [3]最终返回 [1]。但正确答案明明是 [5]。所以先比较再替换不是优化而是正确性的一部分。后面的章节里我还会再展开一次这个坑。2.5 用标准库 heapq 偷懒的正确姿势实际项目里没必要每次都手写堆Python 标准库heapq经过高度优化比自己用class实现更快。等价的实现如下import heapq def topk_largest_heapq(iterable, k): if k 0: return [] it iter(iterable) heap [] try: for _ in range(k): heap.append(next(it)) except StopIteration: return sorted(heap, reverseTrue) heapq.heapify(heap) for x in it: if x heap[0]: heapq.heapreplace(heap, x) return sorted(heap, reverseTrue)这里还是那句heapreplace前必须先比较。标准库不会替你判断“新值是否真的值得替换”判断逻辑必须自己做。3. 复杂度账本、正确性验证与不同输入分布下的实测手写算法最怕“看起来对”所以一定要做正确性验证并且理解复杂度在真实数据下的表现。3.1 复杂度直觉与推导堆方案的时间复杂度由两部分组成建堆 O(K)遍历 N 个元素时每个元素最多一次 O(log K) 的替换。因此最坏情况是每个元素都替换总复杂度 O(N log K)最好情况是后面所有元素都不需要替换退化为 O(N)。空间复杂度很明确堆只保留 K 个元素O(K)。对比全排序的 O(N) 内存在数据量大于内存的前提下堆方案是从“不可能”变成“完全可行”的关键。但有个细节容易被忽略如果 K 接近 N堆方案的复杂度会逼近 O(N log N)空间优势也基本消失此时和排序方案差距不大。所以堆适合 K 远小于 N 的场景K 很大时要用别的方法下一章讲。3.2 用随机数据验证正确性最稳的验证方式是拿堆的结果和全排序的结果直接对比import random data [random.randint(0, 10**6) for _ in range(200_000)] k 10 mine topk_largest(data, k) expected sorted(data, reverseTrue)[:k] assert mine expected强烈建议多覆盖几个边界k1、k len(data)、所有元素都相等、包含大量负数、包含浮点数。这些边界最容易暴露“先比较再替换”之外的 bug比如元素都是相等时是无条件替换还是保留先到的不同需求可能要求不同。3.3 三种顺序输入下的性能差异有一次我拿着相同代码分别跑了升序、降序、随机三种输入得到了很有意思的结果。测试大致这样import time n 1_000_000 k 10_000 cases { 升序: list(range(n)), 降序: list(range(n, 0, -1)), 随机: [random.randint(0, n) for _ in range(n)], } for name, data in cases.items(): start time.perf_counter() topk_largest(data, k) print(name, time.perf_counter() - start)结果通常是降序最快因为填充完 K 个堆元素后后面所有元素都比堆顶小根本不触发替换升序最慢因为每个元素都大于堆顶每次都触发heapreplace和 O(log K) 下沉随机介于两者之间。这个现象告诉我们堆方案的常数依赖“替换频率”而替换频率又依赖数据分布。实际业务里如果数据接近有序堆会非常快如果数据完全逆序堆几乎每步都在做下沉操作。面试时能说出这一层比单纯背复杂度深刻得多。3.4 一个不使用堆的替代当值域很小时用计数桶如果数据是有限值域的整数比如学生成绩 0 到 100求 Top100 根本不需要堆。维护一个长度 100 的频次数组先遍历一遍计数再按分数从高到低取满 K 个。时间复杂度 O(N)空间固定。如果数据是 URL 这类值域很大的 key可以先哈希聚合再对聚合结果做堆。这些方案在特定场景下比堆更优但通用性不如堆。工程选型一定要看数据特征。4. 当 K 本身大到内存放不下分块 TopK 合并与外部排序堆方案有一个硬前提K 个堆元素必须能装进内存。假设 K1 亿每个元素是 Python 的 int实际内存占用远不止 8 字节堆可能要好几 GB这时候单机堆就崩了。怎么处理答案是分治。4.1 分块局部 TopK 合并的正确性把数据分成若干块每块在内存里求该块的局部 TopK然后把所有块的局部 TopK 合并再求一次 TopK。为什么这个做法能保证全局正确用反证法假设全局 TopK 中的某个元素 x它所在的那一块里没有进入局部 TopK。那就说明该块内至少有 K 个元素比 x 大。既然该块内就有 K 个元素比 x 大那么全局必然至少有 K 个元素比 x 大x 怎么可能排进全局 TopK矛盾。因此全局 TopK 的每个元素都一定落在了它所在块的局部 TopK 里。合并所有局部 TopK 再取 TopK不会漏。这个正确性证明非常关键它不只是理论也直接指导了可落地的入门代码def external_topk_largest(file_path, k, chunk_size): candidates [] buffer [] with open(file_path) as f: for raw in f: num int(raw.strip()) buffer.append(num) if len(buffer) chunk_size: candidates.extend(topk_largest(buffer, k)) buffer [] if buffer: candidates.extend(topk_largest(buffer, k)) return topk_largest(candidates, k)这里有个隐患candidates最多会有块数 * k个元素。如果块数很多候选集还是可能超出内存。解决方法是多级合并把candidates再次分块重复同样操作直到候选集能装进内存。这本质上是一个树形归并。4.2 完整外部排序全排序后再取前 K当 K 非常接近 N 时分块局部 TopK 已经没什么优势了因为每块取 K 个几乎等于每块取全部不如直接做完整外部排序分块排序写盘多路归并输出全局有序序列取前 K 个后立刻停止。分块排序阶段可以这样写def write_sorted_blocks(input_path, output_dir, block_size200_000): block_id 0 buffer [] def flush_block(data, path): with open(path, w) as out: out.write(\n.join(map(str, data))) with open(input_path) as f: for line in f: buffer.append(int(line.strip())) if len(buffer) block_size: buffer.sort() flush_block(buffer, f{output_dir}/block_{block_id}.txt) block_id 1 buffer [] if buffer: buffer.sort() flush_block(buffer, f{output_dir}/block_{block_id}.txt) block_id 1 return [f{output_dir}/block_{i}.txt for i in range(block_id)]归并阶段用堆管理每个块的“当前头部”一次能归并多少路取决于能同时打开多少个文件。核心代码如下import heapq def merge_largest_k(block_paths, k): heap [] file_handles [] try: for idx, path in enumerate(block_paths): f open(path) file_handles.append(f) first_line f.readline() if first_line: value int(first_line.strip()) # 块内升序取最大 K 个时头部最小元素取负入堆 heapq.heappush(heap, (-value, idx)) result [] while heap and len(result) k: neg_val, idx heapq.heappop(heap) result.append(-neg_val) line file_handles[idx].readline() if line: heapq.heappush(heap, (-int(line.strip()), idx)) return result finally: for f in file_handles: f.close()这里为什么用负值因为 Python 的heapq是最小堆而我们需要在所有块的当前头部里找最大的那个。把值取负后“最小堆的堆顶”正好对应原值最大的头部。每次弹出一个最大值后从同一个块文件补下一行因为块内是升序的所以补进来的值一定不小于刚弹出的值这符合归并排序的推进逻辑。4.3 两种方案的选型指标分块局部 TopK 合并完整外部排序中间数据量每块最多保留 K 个远小于 N每块全量写盘接近 N能否提前停止可以可以归并输出前 K 即停止实现复杂度低中适用场景K 远小于 NK 接近 N或需要全局有序输出实际中我通常先用分块局部 TopK 合并只有当块数多到候选集放不下或者 K 真的接近 N才开始做完整外部排序。大部分“数据量大于内存”的 TopK 需求用 4.1 的简单方案就够了。5. 多路归并和堆解法里最容易踩的坑这部分全部来自真实排错经验每一条都让项目出过问题要么是结果错误要么是内存炸掉。5.1 无条件heapreplace把大值挤掉的反直觉错误前面提过一次但这值得单独展开。K1 的返例已经足够说明问题数据 [5, 4, 3, 2, 1]如果你在每个元素上直接调用heapreplace堆会一步步变成 [4]、[3]、[2]、[1]最终答案是 1正确答案是 5。原因在于heapreplace的语义是“不管当前堆顶是多少直接把新值放进去再调整”。在求最大 K 个时堆顶是当前候选里的最小值。如果新值比堆顶还小它本来就不该进堆直接替换反而把一个更大的值挤走了。所以if x heap[0]这行判断不仅是为了性能更是正确性的根基。5.2 批量建堆 vs 逐个 push取前 K 个元素时用heapify批量建堆的时间是 O(K)而逐个push是 O(K log K)。当 K 很大时这个差距会非常明显。举个例子K100 万heapify几十毫秒搞定逐个push可能要几百毫秒。看起来不大但在高频调用的数据管道里就是可感知的延迟。用标准库时尤其注意先放列表再heapq.heapify而不是循环heapq.heappush。5.3 读大文件禁用 readlines 和整文件加载“数据量大于内存”的第一铁律绝对不要用readlines()把整个文件读进内存。很多人第一次写大文件处理时都会犯这个错。正确做法是for line in f:文件对象本身就是迭代器每次只读一行。如果一行也非常大比如几十 MB 的单行 JSON需要用固定缓冲区分块读取再按分隔符切分。处理 GB 级文件时IO 模型决定了你的内存峰值而不是你后来用的堆多厉害。5.4 Python int 的内存开销与真实内存预估这是最阴的坑。很多人估算内存时按“每条 8 字节”算但 Python 的int对象实际占 28 字节左右。K1000 万时堆内光是元素对象就约 280 MB再加上列表指针每项 8 字节总计超过 360 MB。如果机器内存只有 256 MB直接 OOM。遇到这类大规模 K优先考虑以下方案之一用array.array(q)或numpy存储数值每个元素只占 8 字节用 C 或 Java 等原生语言实现堆走分块局部 TopK 合并控制每个堆的规模。排错时用sys.getsizeof(obj)逐一检查容器和元素能在上线前发现很多悲剧。5.5 文件描述符用尽与多轮归并外部排序里如果一次性打开几千个块文件操作系统文件描述符直接打满程序会报EMFILE。Linux 默认一般只有 1024即使调ulimit也不建议在一个进程里持有太多个句柄。解决办法是限制单轮归并路数比如每轮最多合并 512 个块生成更大的中间块第二轮继续直到最后只剩一个有序流。注意在finally或with中关闭所有文件否则异常时句柄泄漏会导致后续轮次打不开文件。5.6 空文件、脏数据、重复 key 的边界处理大文件解析时空行、非数字字符串、首尾空格都会导致int(raw.strip())抛异常。工程代码里要加容错比如try-except跳过解析失败的行。另外如果需求要求去重后的 TopK比如同一个 URL 只算一次堆里还要维护一个哈希集合节点比单值复杂得多如果需求允许重复 key 各自占位那就按普通多值处理。这两种语义差异要在算法设计前问清楚。6. 数据分布在多台机器Parallel TopK 的分治思路单机方案讲完后顺便说说分布式场景。其实原理就是第 4 章的局部 TopK 合并只是把“块”换成了“机器”。6.1 局部 TopK 合并为什么能保证全局正确每台机器基于自己分片的数据求局部 TopK再发到中心节点做二次 TopK。这个正确性证明和第 4 章完全一样如果某个元素在它所在的机器里排不进局部 TopK说明这台机器上至少有 K 个元素比它大那么全局至少有 K 个元素比它大它不可能进全局 TopK。但这里有一个非常关键的前提分片之间不能存在“跨片聚合依赖”。简单说如果你求的是纯数值 TopK比如最高温、最大耗时随机分片没问题但如果求的是“按 key 聚合后的 TopK”比如每个 URL 的访问次数 TopK那原始日志不能随机分片否则同一个 URL 散落在几十台机器里任何单机都统计不到完整次数局部 TopK 自然不可信。正确的做法是用哈希把同一个 key 路由到同一台机器每台机器先完成本地聚合再求局部 TopK最后中心合并。这是很多并行计算任务里最容易出错的地方。6.2 中间候选集放不下时的多级归并如果机器数 M 很大每台上传 K 个候选中心节点要处理 MK 个元素。MK 超内存时不要一股脑全拉取应该做树式归并先让若干机器两两合并出这一组的 TopK逐层上报。这和多路归并的逻辑一模一样只是把“文件”换成了“机器”。6.3 采样估算与精确结果的取舍如果只想要一个近似排名比如实时监控里的热 Key 榜完全可以用采样后估算或者用 Count-Min Sketch 这类概率结构速度和成本都低很多。但近似方案无法保证精确也不能应用于“审计”“对账”这类需要精确结果的场景。选型前一定要和需求方确认能不能容忍误差误差范围是多少这些问题往往比算法本身更影响架构设计。最后分享一点个人体会TopK 看起来简单但真正工程化时先确认 K 的量级、再选数据结构、最后验证和压测一步都不能少。我自己的常用套路是直接复制第 2 章的最小堆遇到内存瓶颈就切换成第 4 章的分块策略并用第 3 章的随机数据验证函数兜底。还有一个很实用的小技巧处理超大文件时永远先用 1 万行的小样本压测逻辑确认结果正确后再跑全量。这样排错成本会低很多。希望这篇能让你把“手写 TopK 算法”从面试题变成真正能落地的工程方案。