接到“亿级数据里找Top100”这个需求时，很多人的第一反应是data.sort(reverse=True)[:100]。小数据量跑起来毫无压力，可真到了天级别的日志、埋点、用户行为数据，这行代码能把机器内存直接打满，甚至让进程被系统杀掉。我见过不止一个业务团队在数据量上来之后被这个朴素方案坑过，最后才换成堆方案，几亿条数据单机也能在几十秒内跑完。这题的解法核心说穿了就是一个数据结构：堆。
这篇文章我把整套做法拆开讲：为什么堆在这个场景下是正解、Python 里 heapq 的实际用法、亿级数据下的性能实测思路、以及真实业务里我会怎么处理动态数据流、分组 TopK 这些变种。适合刚接触算法优化的开发者，也适合正在为线上大数据任务发愁的工程师。
1. 亿级数据TopK的选型思路
1.1 朴素方案为什么在大数据下挂掉
全量排序的思路在数学上没有任何错误，但工程上代价极其高昂。排序的时间复杂度是 O(N log N)，空间复杂度是 O(N)。这里的 O(N) 不是抽象复杂度，是实打实的内存占用。Python 里一个整数对象大约占 28 字节，1 亿个整数光数据本身就要 2.8 GB，再加上列表里每个元素的 8 字节指针引用，总占用轻松超过 3.6 GB。如果数据是字符串、字典或自定义对象，内存消耗还会成倍上涨。
更隐蔽的问题在于，全量排序做了一堆无用功。你要的只有 100 个最大值，但排序会把 N-100 个“一定进不了Top100”的元素也排得整整齐齐。数据量越大，浪费越明显。等到数据大到内存放不下，就得引入外部排序、分片处理、数据库临时表这些额外设施，复杂度一下子就上去了。
核心矛盾就在这里：N 很大，K 很小，中间隔着几个数量级。这种场景下，算法选型的唯一原则就是“别碰不该碰的数据”。
1.2 堆为什么是正解
堆能在 O(N log K) 时间内解决 TopK，空间只有 O(K)。数据量为 10 亿时，log K（K=100）约等于 6.64，而 log N 约等于 29.9，差出 4 倍多；更关键的是空间从“存下所有数据”变成“只存 100 个元素”。
堆解决 TopK 的思路非常直白：维护一个大小为 K 的小顶堆，堆顶就是当前候选集合里最小的值。遍历所有数据时，如果新元素比堆顶大，就把堆顶挤出去，把新元素放进来。遍历结束后，堆里剩下的就是全局最大的 K 个元素。这个过程天然支持流式处理，不需要一次性加载全部数据，数据来一条看一条，非常适合日志、埋点这种源源不断的场景。
Python 标准库的 heapq 模块把这个数据结构封装得很顺手。底层是数组实现的完全二叉树，heappush 和 heappop 都是 O(log K)，heapify 可以线上性时间把一个数组变成合法堆。我后面会详细说这些方法的正确用法，以及几个容易踩的坑。
1.3 其他方案的适用边界
堆不是唯一解，面试或实际选型时经常拿来做对比的有 QuickSelect（快速选择）和计数排序。
QuickSelect 的平均时间复杂度是 O(N)，比堆的理论复杂度更好，但它有两个工程硬伤。第一，最坏情况退化到 O(N²)，虽然随机化后概率很低，但线上系统很难接受这种不确定性。第二，它需要把所有数据一次性放进内存做原地划分，数据规模一大照样卡在内存上。所以 QuickSelect 比较适合 K 接近 N、内存又放得下的场景，比如从 10 万条数据里取前 5000 名。
计数排序在数据范围很小的场景下很香，比如分数固定是 0 到 100 的整数，开个 101 长度的桶扫一遍就完事，时间复杂度 O(N)。但业务数据很少有这种好脾气，分数、金额、时间戳这些数字范围往往极大，稀疏到没法开桶。
用一句话概括选型逻辑：内存放不下全部数据时优先找堆和流式方案；内存放得下且 K 较大时优先 QuickSelect；数据范围小时优先计数排序。大多数“亿级数据求Top100”的题目，堆都是最稳健、最不容易翻车的答案。
2. heapq基础与核心TopK实现
2.1 heapq模块的关键操作
Python 的 heapq 模块没有单独暴露“堆”这个类，而是提供一组操作列表的函数。列表本身就是一个完全二叉树的内存布局：索引 i 的父节点在 (i-1)//2，左子节点在 2i+1，右子节点在 2i+2。heapq 围绕这个关系做上浮和下沉操作。
日常用到的就四个：
heapify(heap)：原地把列表整理成堆，线性时间复杂度heappush(heap, item)：末尾插入元素并做上浮调整heappop(heap)：弹出堆顶最小元素，并把最后一个元素放到堆顶做下沉调整heapreplace(heap, item)：先弹出堆顶，再插入新元素，比先 pop 再 push 少一步调整过程
还有一个容易被忽略的组合操作：heappushpop(heap, item)，先把新元素 push 进去，再弹出最小值。在“不断加入元素但只保留最大 K 个”的场景里，这两个函数配合起来非常顺手。
2.2 只提供小顶堆？取负技巧来凑
一个让很多新手困惑的点：heapq 默认是小顶堆，堆顶永远是最小值。可是求最大 TopK 时，我希望堆顶是“当前候选里最小的那个”，这样新元素只需要和堆顶比较，比堆顶大就替换。你会发现小顶堆正好就是这个需求，所以求最大 TopK 直接用默认小顶堆，逻辑上完全匹配，不需要任何取巧。
真正需要变通的是求最小 TopK。这时候我希望候选集合里“最大”的据点在堆顶，新元素只要比堆顶小就替换，堆顶需要变成最大值。但 heapq 没有天然的 max-heap，标准做法是存-x：所有元素取负后，原本最大的值变成最小的负数值，站在堆顶。最后返回时再把结果取负还原。
这里有个特别容易踩的坑：数据里本身有负数时，取负之后正负交换，比较关系完全反转，但堆程序只认“数字越小越靠上”的逻辑，所以结果依然正确。关键在于你要始终保持心理预期——堆里存的是负的，最后千万别忘了反转回来，否则输出的 TopK 会带着负号，而且顺序完全错乱。
2.3 手写TopK代码与nlargest的取舍
手写一个求最大 TopK 的函数并不复杂：
import heapq def top_k_max(iterable, k): heap = [] for x in iterable: if len(heap) < k: heapq.heappush(heap, x) elif x > heap[0]: heapq.heapreplace(heap, x) return heap注意这里用heapreplace而不是先heappop再heappush，一步到位，省一次下沉调整。数据流很大时，这一个小小的区别能省下不少开销。堆内元素数量始终不超过 k，所以整个遍历过程的空间占用是 O(k)。
Python 标准库其实也提供了现成的heapq.nlargest(100, data)和heapq.nsmallest(100, data)。源码层面，它会根据 n 和序列长度的关系做切换：n 远小于序列长度时用堆；n 和序列长度接近时直接用sorted，因为排序的优势这时候更明显。所以如果数据已经全部在内存里、且 n 很小，用系统的nlargest就够了，连手写都省了。
但注意一个隐藏问题：nlargest为了返回有序结果，内部会维护一个大顶堆（或者反过来），最后还要排序。如果只需要找出 Top100 而不在意顺序，手写方案能省掉最后的排序；如果需要按分数从高到低展示，nlargest返回的结果已经排好序，直接用更省事。两种方案各有取舍，我在 3.2 的实测里给过对比数据，实现差异很小，按场景选就行。
def top_k_min_with_negation(iterable, k): heap = [] for x in iterable: if len(heap) < k: heapq.heappush(heap, -x) elif -x > heap[0]: heapq.heapreplace(heap, -x) return [-v for v in heap]2.4 堆里存对象怎么办：元组堆的排序逻辑
实际业务里的数据很少是裸数字，更多是“用户ID + 分数”这样的结构。如果直接把自定义对象塞进堆，Python 会比较对象本身，没有实现__lt__的话直接报类型错误。
常见做法是存元组(score, user_id)。元组比较时先比第一个元素，分数相同才比第二个元素，这样可以避免比较整个对象。如果你只想按分数排序，但分数可能相同，第二个元素用 user_id 或一个自增序号就能打破平局，还避免了自定义对象在比较时引入的额外开销。
我见过有人为了解决“对象不能比较”的问题，给对象实现了好几个魔法方法，性能也没好到哪去。其实用一个二元组就解决了，这也是堆操作里最实用的小技巧。
3. 模拟亿级数据与性能实测
3.1 用生成器模拟流式数据
先放下理论，看看实际跑出来的效果。正式测试前要先解决“怎么造亿级数据”的问题。直接data = list(range(100_000_000))会把机器内存打爆，Python 的大整数对象实在太占空间。
更稳的做法是用生成器。它在每次迭代时才产生一个数，整个过程中内存里只存在一个当前值，堆里最多存 100 个候选值。这样既模拟了真实流式场景，又规避了测试时内存不够的问题：
def data_stream(total): # 伪随机生成器，保证测试可复现 import random rng = random.Random(42) for _ in range(total): yield rng.randint(0, 10**18)这里有个细节：生成器每次只产出下一个值，调用方不能在遍历中途“回头看”，所以它天然逼迫你按流式方式处理。如果要从文件里逐行读，也是一样的结构——打开文件、逐行解析、逐条进堆，读满100个候选之后就开始做淘汰。
3.2 三种方案的实测对比
我在一台普通开发机上，用 1000 万条随机整数做了对比。结果大约是这样：
| 方案 | 时间复杂度 | 内存 | 实测耗时（相对） |
|---|---|---|---|
全量sorted后切片 | O(N log N) | O(N) | 约 6 ～ 8 秒 |
heapq.nlargest(100, data) | O(N log K) | O(N) | 约 1.5 秒 |
| 手写堆遍历 + heapreplace | O(N log K) | O(K) | 约 1.2 秒 |
注意一个容易被忽略的点：heapq.nlargest需要把整个数据序列传给函数，如果数据已经是一个列表，额外的内存占用来自这个列表本身；如果传生成器，内部会先把生成器“拉完”才能确定长度吗？并不会，nlargest内部实现其实是一次性迭代的，但它在开头会先判断 n 和整个可迭代对象长度关系，如果要保证这个判断，对生成器它会先尝试用len()，生成器没有len()时会走另一种分支，所以传生成器也安全。不过提前知道数据规模的场景，直接传列表更直观。
手写堆的内存占用是 O(K)，这是它和nlargest的最大区别。数据量到 10 亿级别时，sorted方案基本不可用——先不说内存，光把 10 亿个 Python 对象生成出来就够呛了。而堆方案即使数据是分布式存的多份文件，只要按流式读入，内存占用始终只有 100 个元素加一个迭代器。
3.3 数据规模扩大后会发生什么
1000 万条和 1 亿条差了一个数量级。我在 1 亿条时测了手写堆方案，耗时约 12 秒，内存峰值始终稳定在几 MB 级别。sorted方案在 1 亿条时还没跑完内存就已经报警了，进程直接被系统 kill。
时间增长也验证了复杂度分析：N 翻 10 倍，log K 不变，所以堆方案的时间接近线性增长；排序方案是 N log N，N 每翻 10 倍，耗时大约变成原来的 20 倍左右。两者差距会越来越大，这也是为什么“亿级数据Top100”这道题几乎不会有人推荐排序当答案。
这里顺带说一句，如果数据真实存在磁盘上的文本文件里，逐行读取 + 堆维护比一次性内存加载再算要靠谱得多。几十 GB 的文本，读一行处理一行，内存占用永远是常数级别。我在某数据平台的埋点日志处理中就是这么干的，最初用全量排序导致节点内存报警，换成堆之后，同样的任务从几小时降到十几分钟。
4. 真实场景中的TopK变种与工程化
4.1 动态数据流上的TopK追踪
线上业务里数据不是静止的一坨，而是源源不断的流。典型场景是实时推荐系统对用户行为打分，分数一直在更新，产品要随时能看到当前总榜前 100 名。
这种情况相当于把堆从一次性任务变成长期维护的在线结构。核心逻辑不变，只是要包一个类：
class TopKTracker: def __init__(self, k): self.k = k self.heap = [] def add(self, value): if len(self.heap) < self.k: heapq.heappush(self.heap, value) elif value > self.heap[0]: heapq.heapreplace(self.heap, value) def top(self): return sorted(self.heap, reverse=True)add方法每次 O(log K)，即使每秒进来百万级事件，只要 K 不大，开销也能扛住。top方法在需要展示时才排一次序，不影响增量更新的性能。如果把value换成(score, user_id)元组，还能顺带追踪是谁排在前列。
这里注意一个违背直觉的细节：动态场景下，如果某个已经在堆里的元素的分数增加了，堆不会自动重新调整。需要额外维护“元素 → 堆中位置”的映射，做位置更新和上浮操作。这属于工程化改造，复杂度上去一个档次。如果业务上允许，更简单的方式是定期用当前窗口内的数据重建堆，比如每分钟用最近一小时的数据重建一次 Top100，而不是实时逐条更新。
4.2 分组TopK：多轮堆合并
实际业务里经常要的不是全局 Top100，而是每个分组各取 Top100，比如每个商品类目下销量最高的 100 个，或者每个省份活跃度最高的 100 个用户。
最简单粗暴的方式是遍历数据，为每个分组维护一个小顶堆。伪代码这样写：
from collections import defaultdict group_heaps = defaultdict(list) for group_id, score in stream: heap = group_heaps[group_id] if len(heap) < K: heapq.heappush(heap, score) elif score > heap[0]: heapq.heapreplace(heap, score)这种方式内存占用是“分组数 × K”，分组特别多时总有压力。另一个思路是先用一轮遍历把数据按分组写到临时分片文件里，再对每个分片单独做 TopK。这样任何一个时刻内存里都只有一个分组的数据，适合分组数量多、单组数据量又很大的场景。
更进一步的优化是“两阶段堆”：第一阶段，每个分组各自用 K 大小的小顶堆筛一遍，得到每组 TopK 候选；第二阶段，把所有候选取出来再做一次全局排序/堆合并，得到每组最终的 TopK。这个方案适合数据量特别大、不能全部放内存的离散场景，本质上是把一次大问题拆成多次小堆操作。
4.3 分布到多个分片时怎么做
数据如果本来就是分布在不同节点或不同文件上的，不要先把所有数据汇总再算，那样等于放弃了分布式带来的好处。正确姿势是“分片局部 TopK + 全局合并”。
为什么这个策略能保证正确性？反证法：如果某个元素不是它所在分片的前 100 名，那说明它所在分片至少有 100 个元素比它大。全局 Top100 只取 100 个元素，那 100 个比它大的元素里至少有一个会排在它前面，所以这个元素永远不可能进入全局前 100。结论：每个分片只需要贡献自己的前 100 名，全局 Top100 一定在这“分片数 × 100”个元素里。
这个性质让分布式 TopK 变得非常简单：每个工作节点各自跑一个堆，把 100 个结果传给汇总节点，汇总节点再维护一个大大小为“节点数 × 100”的堆，或者直接排序后截断。数据规模再大，汇总阶段的量也只是“节点数 × K”，完全可控。这也是我在实际团队里最常用的方案，配合分批读取和堆合并，单机到多机都能平滑扩展。
4.4 从Top100到双堆技巧：滑动窗口极值
双堆是堆技巧里比较进阶的一个场景：维护一个最大堆和一个最小堆，动态跟踪数据流的中位数，或者滑动窗口的最大/最小值。思路是让两个堆各存一半数据，始终保持“左堆顶 < 右堆顶”，中位数就可以 O(1) 时间拿到，插入的复杂度是 O(log N)。
Python 里累最大堆只能用取负的方式，因为这个“下标准库只有一个方向”的限制，双堆代码看着总有那么点别扭。但如果你已经吃透了 TopK 里的取负技巧，双堆的代码无非是再套一层逻辑。我在做实时监控指标时用过这类结构：同时跟踪当前窗口的最大值、最小值、中位数和 TopK，一个数据结构全搞定。从 TopK 入手，双堆和滑动窗口极值都是很自然的延伸。
5. 常见问题与避坑指南
5.1 堆大小和堆顶判断的易错点
TopK 核心循环里有三个判断分支：堆没填满时直接 push；堆填满后新元素大于堆顶时替换；小于等于堆顶时忽略。很多人会把前两个分支写反，或者漏掉“堆没填满”的初始化阶段。
以最经典的错误为例：
for x in data: if x > heap[0]: # 堆还没填满时这里会报 index error heapq.heapreplace(heap, x)堆为空时访问heap[0]直接抛异常；堆不满 100 个元素时，替换逻辑也会把堆搞得乱七八糟。正确的写法一定先把堆填满到 K 个，再接淘汰逻辑。另外注意边界条件：新元素等于堆顶时替换不替换都行，不影响最终集合，但会影响排序后的顺序。如果业务要求保留“先到达”的数据，等号时不替换更公平。
5.2 取负技巧的坑
求最小 TopK 用最大堆，Python 里实现最大堆靠取负。这个转折容易在三个地方出问题：
第一，初始化堆时直接存-x，最后返回时要取负还原，漏一步返回的全是负数。第二，如果 x 是浮点数或字符串，取负可能不支持或者行为不符合预期，字符串根本不能取负，只能用元组里加一个比较键的方式模拟。第三，数据里有 0 或负数时，取负后正负号会反转，但你最终关心的是原值，只要最后记得还原，堆内部比较逻辑不受影响。
我在 review 同事代码时见过一个很隐蔽的 bug：他求最大 TopK 时没取负，但因为堆里存的是自定义对象，叠了某个魔法方法，居然跑出了“看起来对”的结果。这种隐性依赖时序一乱就容易翻车，不如老老实实用元组，把分数放第一个字段。
5.3 数据里有重复值怎么办
重复值在堆里不会引发崩溃，但会带来一个常见误区：TopK 要“100个不同的元素”还是“值可以重复的100个位置”。如果业务要求去重后的 Top100，单纯用堆没法做到，因为堆只会挑最大的 100 个值，不会自动去重。
处理方式分两步。第一步：在数据进入堆之前用一个哈希集合过滤重复值，或者用消息的去重逻辑。第二步：对过滤后的数据流跑堆。哈希集合本身也有内存开销，所以更精细的做法是双阶段——先粗筛出可能进 TopK 的大值区间，再在区间内精确去重，避免对几亿条重复数据都维护一个巨大的 set。
如果业务只关心“前100名成绩单”，重复值不影响排名，那直接堆就行，不用加额外逻辑。
5.4 heapreplace 和 heappushpop 该怎么选
替换逻辑有两个选择：heapreplace先弹出最小再插入，新元素一定进堆；heappushpop先插入再弹出，新元素可能进堆也可能直接被弹出去。看代码细节会发现，heapreplace更适合“已经确认新元素大于堆顶，需要替换”的场景，heappushpop适合“不确定要不要替换，直接丢进去让堆自己决定”的场景。
从一次操作的效果看，两者时间复杂度都是 O(log K)，但内部路径不同。heapreplace是直接从堆顶下沉，heappushpop是先上浮再下沉。数据量极大时，实测差异其实很小，但用语义更明确的那个，代码的可读性会好很多：新元素比堆顶大才替换，就用heapreplace；无脑塞入让堆决定，就用heappushpop。
5.5 性能调优的几个方向和边界
CPython 的 GIL 决定了多线程对纯计算任务帮助不大。遇到超大 TopK 任务，更实际的方向是分片读取配合多进程，或者直接上分布式框架，但核心的每个分片内仍然是堆逻辑。
微调方面，能批量整理时优先heapify而不是逐个heappush。比如你已经收集了 K 个元素，准备建堆，直接heapify(collect_list)是 O(K)，而逐个 push 是 O(K log K)，差距在 K 较大时很明显。常规 TopK 的 K=100 差距不大，但既然只差一行代码，用 heapify 没坏处。
数据倾斜也会影响实际效果。某个分组的数据量远大于其他分组，堆的淘汰竞争会更激烈，但单次操作仍然是 log K。如果想进一步压榨性能，可以对每个分组采用“采样估计阈值 + 过滤后再堆”的思路：先随机抽一部分数据估计出 TopK 的大致分数阈值，正式遍历时小于阈值的元素直接丢弃，减少堆的比较次数。这个技巧适合数据量极大、单条数据又是一行解析成本很高的场景，能省掉不少无意义的比较。
6. 经验总结与扩展
6.1 我踩过的几个坑
我先说一个印象很深的坑。某次做埋点日志 Top100，数据从文件逐行读取，我一开始没注意文件里有些行是脏数据，直接取了字符串转数字。某一行解析失败抛异常后，整个任务崩了，前面几十 GB 的堆状态全废。后来我在解析和进堆之间加了一层校验和容错，只跳过异常行，不中断整个流程。这个习惯后来救了我好几次。
另一个坑是关于内存的误判。我以为堆只存 100 个元素就一定稳，结果发现数据的每条记录还有关联上下文，如果把这个上下文也塞进堆，内存就不是 100 个数字而是 100 个完整对象。实际业务里我一般只把必要的排序键和唯一标识放堆里，详情数据留在外部存储，要用时再按标识去取，堆里绝不放大对象。
还有一次，我在动态流场景里忘了处理“已经在堆里的元素分数更新”的情况。某个用户分数暴涨，按理说应该进 Top100，但因为堆里已经存了他的旧分数，新分数只会被当作普通新元素和堆顶比较。旧分数还在堆里占了位置，导致最终结果不准。后来我在流式场景里改用定期重建窗口内的 TopK，彻底绕开了这个问题。
6.2 堆之外还能做什么
TopK 这个核心思路能延伸出很多变体。除了前面说的双堆中位数、窗口极值，还有一个常见扩展是“文件级别的多路归并”：每个文件内部先排序成有序序列，然后多个文件轮流从头部取元素做合并，本质上还是堆在挑最大/最小。大数据生态里的排序、优先队列、任务调度，几乎处处能看到堆的影子。
面试里问到 TopK，可以秀的深层内容很多——复杂度推导、内存模型、流式支持、分布式扩展、重复值去重、正确性证明，每一个都能展开聊。但核心都是从“维护大小为 K 的堆”这一个动作发散出去的。把这个动作练透，比背一堆八股文有用得多。
最后说点实际的收获。我处理过某数据平台每天几亿条埋点日志的 Top100 任务，最初用全量排序导致节点内存报警，换成手写堆 + 分片合并之后，单机任务从几小时缩短到十几分钟。这个收益不是算法炫技，而是选型正确、数据结构和真实场景匹配带来的红利。如果你也碰到类似需求，先别急着写大排序，停一秒，想想能不能用一个小小的堆解决问题。
