400 万条 ES 数据全量拉取与实时评分：我们是怎么把 2 小时压到 5 分钟的

起因

线上有个 Elasticsearch 索引 sdb_indexv4，存了大概 400 多万条记录。业务需求不复杂——每条数据有个指标值 x_age，需要算出它在全量数据中的百分位排名，公式很直白：

1	1score = 100 * (比 x_age 小的数据条数) / 总条数

听起来就是个数数的事。但数据量一上来，问题就一个接一个。

最初的做法，和它是怎么崩的

第一版代码写得很”理所当然”：

1# 用 scroll 拉全量数据，每次取 1000 条
2resp = es.scroll(scroll_id=scroll_id, scroll='5m')
3for hit in resp['hits']['hits']:
4    all_docs.append(hit['_source'])  # 攒起来
5
6# 排序，然后对每条数据计算排名
7sorted_values = sorted([d['x_age'] for d in all_docs])
8for doc in all_docs:
9    rank = bisect_right(sorted_values, doc['x_age'])
10    score = 100 * rank / len(sorted_values)

跑了一下，三个问题同时炸了：

内存直接打爆。400 万条 _source 全塞进一个 list，单机内存吃紧，跑到后半段 OOM 了。
Scroll 上下文超时。中间网络抖了一下，scroll_id 失效，400 万条从头再来。
太慢。scroll 本身要维护快照上下文，越往后越慢，跑完一次全量将近两小时。

这不是优化能解决的问题，是方案本身就不对。

重新想这件事

坐下来理了一下，这个任务本质上就两件事：

离线阶段：把 400 万条数据过一遍，构建每个指标的全局有序数组。
在线阶段：来一条数据，在有序数组里二分查找，算出百分位。

在线阶段其实没什么好优化的——bisect_right 在 400 万长度的数组里查一次也就二十几次比较，微秒级的事。瓶颈全在离线阶段：怎么又快又稳地把 400 万条数据从 ES 里搬出来，构建好有序数组。

所以问题收敛成：怎么可靠地、低内存地、可中断恢复地从 ES 拉取 400 万条数据。

第一件事：扔掉 Scroll，换成 search_after

Scroll API 的设计初衷是做短时间内的批量导出，它会在服务端维护一个快照上下文。这个上下文是有状态的——占内存、有超时时间、阻止段合并。400 万条数据要拉很久，scroll 上下文扛不住。

search_after 是无状态的。原理很简单：每次查询带上一页最后一条数据的排序值，ES 根据这个值定位下一页的起点。服务端不保留任何上下文。

1// 第一页
2GET sdb_indexv4/_search
3{
4  "size": 2000,
5  "query": { "match_all": {} },
6  "sort": [{ "_id": "asc" }]
7}
8
9// 后续页，带上上一页最后一条的 sort 值
10GET sdb_indexv4/_search
11{
12  "size": 2000,
13  "query": { "match_all": {} },
14  "sort": [{ "_id": "asc" }],
15  "search_after": ["last_doc_id_from_previous_page"]
16}

切到 search_after 之后，最大的好处不是速度，是心安。不用惦记 scroll 上下文过没过期，不用在低峰期赶时间跑完。它就是一个普通的查询，失败了重试当前页就行。

第二件事：别攒数据，边拉边处理

之前的代码把所有 _source 攒到 list 里再处理，400 万条 × 每条大概 1~2KB，内存直接吃掉好几个 G。

改法很朴素——用生成器，数据来了就处理，处理完就丢掉：

1def fetch_all():
2    search_after = None
3    while True:
4        body = {
5            "size": 2000,
6            "query": {"match_all": {}},
7            "sort": [{"_id": "asc"}],
8            "_source": ["x_age", "visit", "download", "ref", "date"]
9        }
10        if search_after:
11            body["search_after"] = search_after
12
13        resp = es.search(index="sdb_indexv4", body=body)
14        hits = resp["hits"]["hits"]
15        if not hits:
16            break
17
18        for hit in hits:
19            yield hit["_source"]
20
21        search_after = hits[-1]["sort"]

注意 _source 里只取了评估需要的五个字段，把那些大段文本、嵌套对象全丢了。光这一步，网络传输体积就少了七成以上。

消费端也是流式的：

1import gzip
2
3values = []
4with gzip.open("export.jsonl.gz", "wt") as f:
5    for doc in fetch_all():
6        values.append(doc["x_age"])       # 只攒一个 float，不是整条文档
7        f.write(json.dumps(doc) + "\n")   # 边拉边写，压缩落盘
8
9values.sort()
10# values 就是全局有序数组，400 万个 float 也就几十 MB

内存里始终只有两样东西：当前这一批 2000 条文档，和一个不断增长的 float 数组。跑完全程内存峰值不到 200MB。

第三件事：加一层 Redis，让进程可以”死而复生”

流式拉取解决了内存问题，但还有一个隐患：进程要是挂了怎么办？400 万条跑了 380 万，突然被 kill 了，从头来？

加一层 Redis 记录进度：

1PROGRESS_KEY = "export:sdb_indexv4:progress"
2
3def fetch_all():
4    search_after = redis.get(PROGRESS_KEY)  # 恢复点
5    processed = 0
6
7    while True:
8        # ... 构造查询 ...
9        resp = es.search(index="sdb_indexv4", body=body)
10        hits = resp["hits"]["hits"]
11        if not hits:
12            break
13
14        for hit in hits:
15            yield hit["_source"]
16
17        search_after = hits[-1]["sort"]
18        processed += len(hits)
19
20        # 每批次更新进度
21        redis.set(PROGRESS_KEY, json.dumps(search_after), ex=86400)
22
23    redis.delete(PROGRESS_KEY)  # 跑完了，清理

重启之后，进程读 Redis 拿到上次的游标，接着跑。不会重复，不会遗漏。

这个设计还有个额外好处——多个 Worker 可以共享同一份进度和缓存表。比如按日期范围拆成子任务并行跑，最后合并有序数组，速度可以再快一倍。

第四件事：构建有序数组，在线评估降维到 O(log N)

离线阶段跑完之后，我们手里有每个指标的有序数组，比如 sorted_x_age，400 万个 float，内存里占几十 MB。把它加载到 Redis 或者直接用共享内存：

1# 在线评估
2def calc_score(x, sorted_array, total):
3    rank = bisect_right(sorted_array, x)
4    return 100 * rank / total

来一条数据，x_age = 37，调一次 calc_score，二十几次内存比较，微秒级返回结果。

从 O(N) 的全库扫描，降到了 O(log N) 的二分查找。400 万条数据，log₂(4000000) ≈ 22。22 次比较，完事。

最终方案一览

把上面四件事串起来，整个流程是这样的：

1┌─────────────────────────────────────────────────────────┐
2│                    离线阶段（每日凌晨）                     │
3│                                                         │
4│  ES ──search_after──▶ Generator ──┬──▶ 提取指标值         │
5│       (无状态分页)      (流式)      │      ↓               │
6│                                   │   排序 → sorted_x     │
7│                                   │      ↓               │
8│                                   │   加载到 Redis/内存    │
9│                                   │                      │
10│                                   └──▶ gzip 压缩落盘      │
11│                                                         │
12│  Redis: 记录 search_after 游标，支持断点续传               │
13└─────────────────────────────────────────────────────────┘
14
15┌─────────────────────────────────────────────────────────┐
16│                    在线阶段（实时请求）                     │
17│                                                         │
18│  请求携带 x_age = 37                                     │
19│       ↓                                                 │
20│  bisect_right(sorted_x, 37) → rank                      │
21│       ↓                                                 │
22│  score = 100 * rank / N                                 │
23│       ↓                                                 │
24│  返回结果（微秒级）                                        │
25└─────────────────────────────────────────────────────────┘

改造前后对比

	改造前	改造后
全量拉取耗时	~2 小时（经常中断）	~5 分钟
内存峰值	4GB+，经常 OOM	< 200MB
中断恢复	从头开始	Redis 断点续传
单次评分耗时	秒级（扫全库）	微秒级（二分查找）
ES 压力	scroll 锁资源	普通查询，无状态

一些踩过的坑

1. search_after 的排序字段必须有唯一性。 如果你用 x_age 排序，大量相同值会导致分页结果重复或遗漏。用 _id 做排序字段最稳，它是全局唯一的。如果需要按 x_age 排序，可以加 _id 作为第二排序字段："sort": [{"x_age": "asc"}, {"_id": "asc"}]。

2. Redis 存游标时注意序列化。 search_after 返回的 sort 值是个数组，可能是 [1234, "doc_id"] 这种混合类型。存 Redis 前 JSON 序列化，取出来反序列化，别直接 str() 转换。

3. 压缩写入时注意 flush。 gzip.open 默认有缓冲区，如果进程意外退出，缓冲区里的数据会丢。可以定期 f.flush() 和 os.fsync()，或者接受最后几百条数据的丢失（反正有断点续传）。

4. 有序数组的刷新策略。 我们是按日刷新，凌晨跑一次全量构建。如果数据更新很频繁，可以考虑增量方案——新数据插入有序数组（用 insort），但要注意 400 万条的数组 insort 是 O(N) 的，频繁插入不合适。折中方案是每小时全量重建一次，或者用两次数组（历史有序 + 新增无序），查询时两边各查一次再合并。

写在最后

回过头看，这个方案没有什么特别高深的技术。search_after、bisect_right、Redis 存状态，都是现成的东西。关键在于把事情拆开想：

离线能做的事，别放到在线做。
能用无状态解决的，别用有状态。
内存里只放必要的东西。
进程随时可能死，设计时假设它下一秒就会死。

400 万条数据不算大，但如果方案不对，它也够你喝一壶的。希望这篇文章能帮到遇到类似问题的朋友。