企业官网建设流程全解析

凌晨两点半，任务跑到 73% 的时候，目标集群的协调节点因为一次批量写入的堆积被拖到几乎无响应，客户端连接池被占满，脚本抛出超时异常直接退出。第二天早上你打开监控，发现已经写进去的 1.2 亿条文档既不知道该不该删，也不知道从哪里接着跑——这是几乎所有做 Elasticsearch 数据迁移的人都遇到过的场景。全量重跑一次的成本往往不是几分钟，而是十几个小时加一整夜的机器资源，更糟的是重跑过程中源集群还在变化，你会开始怀疑是不是永远跑不完。

这篇内容聊的就是怎么把这种"一次性梭哈"的迁移方式，改造成可以中断、可以恢复、可以反复验证的断点续传模式。核心关键词是Elasticsearch 数据迁移和断点续传，我会从"为什么常规的_reindex救不了你"讲到"游标到底该存在磁盘上还是数据库里"，再到一份可以直接改改就跑的脚本骨架。适合已经做过至少一次 ES 迁移、踩过 scroll 超时或者 reindex 任务消失的读者，也适合正准备把某个大索引搬到新集群、想先把方案想清楚再动手的人。

把这件事说清楚之前，得先给个结论：断点续传的难点从来不在"传"，而在"断"。传是 bulk 写入，是吞吐调优，这部分的资料满地都是；断是状态管理，是游标语义，是怎么保证同一批数据跑两遍不会出问题——这部分才是真正决定你的迁移方案能不能在生产上用的东西。

1. 迁移跑到一半断掉时，真正让人难受的不是重跑，而是不知道从哪接着跑

很多人第一次做迁移会有一个朴素的直觉：反正文档写入是按_id覆盖的，重跑一遍不就完了，反正幂等。这个直觉在数据量小的时候是对的，但在亿级索引上会撞到三堵墙。

1.1 reindex 不是事务，它没有回滚也没有续跑

_reindex是大多数人第一个想到的工具，写个 body 提交上去，加个wait_for_completion=false拿一个 task id，然后轮询_tasks接口看进度。问题在于，这个 task 是运行在集群内存里的，一旦协调节点重启、任务被 kill、或者reindex.remote.whitelist那边的远端连接抖断，这个 task 就从_tasks列表里消失了，你手里那个 id 变成一个查不到的死号。

翻一下_reindex的官方文档你会发现，它压根没提供"resume"这个选项。你能做的只有重新提交一次，然后它从第一个分片的第一条文档开始重新扫。虽然因为_id相同，已经写过的文档会被覆盖而不是重复，看起来"没坏"，但代价是所有已经完成的扫描工作全部归零。1 亿条文档跑了 8 小时，跑到 90% 断掉，重跑还是 8 小时，而不是 0.8 小时。

更隐蔽的一点是，重新提交_reindex时如果源索引在这期间有新写入，你没法保证两次扫描看到的是同一个数据快照。第二次扫描会把这些新文档也带上，看起来是好事，但它同时也可能漏掉那些在第一次扫描时已经读过、在第二次扫描时因为段合并被移动位置的文档——概率很低，但在大索引上是真实存在的。所以"重跑=幂等"这个假设，只在源索引完全静止的前提下成立。

1.2 断点续传的成本账：全量重跑和续跑差多少

我拿一个真实的量级算过一笔账。假设源索引 2 亿条文档、平均 1.5KB、总共约 300GB，目标集群单批 bulk 5000 条、8 个并发切片，实测吞吐大概在 8000-12000 条/秒这个区间。跑完整轮大概 5-7 小时。

这里有个关键数字：中断发生在什么位置，决定了重跑亏多少。如果是在 20% 断的，重跑损失也就是多跑一小时；如果是在 90% 断的，重跑等于白干一整晚。而现实里，中断往往更容易发生在后半段——因为越往后，目标索引的段越多、写入放大越明显、refresh 压力越大，出问题的概率反而是上升的。

提示：迁移任务的失败概率和时间不是均匀分布的，是递增的。这个特性决定了"能不能续跑"比"跑得多快"更重要。

除了时间成本，还有一笔经常被忽略的账：源集群的查询压力。每多跑一次全量扫描，就多消耗一次源集群的 IO 和 CPU。如果源集群是承载线上业务的，多跑一轮就意味着多一轮影响业务的风险，这个成本比机器时长贵得多。

1.3 什么场景下才值得上断点续传

不是所有迁移都需要断点续传。我一般用下面这张表来快速判断：

这里最后一行值得多说一句。如果两端集群版本兼容、能拿到快照仓库的读写权限，快照恢复（snapshot/restore）是唯一不需要考虑断点续传的方案，因为它是文件级别的复制，断了重来也只是重传增量段。但现实里很多迁移是跨版本、跨云、跨账号的，快照仓库未必通，这时候自研的续传脚本才有存在价值。

2. 四条迁移链路的续传能力盘点：reindex、logstash、快照、自研脚本

在动手写脚本之前，值得先把手上这几条链路各自的"续传语义"摸清楚。很多人卡住不是因为不会写代码，而是选了一条天生不支持续传的链路，然后花两周时间去给它打补丁。

2.1_reindex与 remote reindex 的能力边界

_reindex的能力其实比很多人以为的强，它支持slices自动切片、支持script做字段转换、支持size控制批大小、支持requests_per_second做限流，还支持conflicts=proceed跳过版本冲突。这些参数组合起来，其实已经能应付大部分中等规模的迁移。

但它的边界也很清楚：

• 任务状态不落盘。task 存在集群状态里，节点重启就没了。
• 不支持指定起始位置。没有任何参数能告诉它"从_id大于 xxx 的地方开始"。
• slices切分后各切片独立推进，但对外只暴露一个聚合进度，你没法知道哪个切片跑完了、哪个卡住了。

最后一个问题在实际操作里特别恶心。_tasks接口返回的进度里有个slices数组，但它给的是每个切片的total和done，任务一旦重启，这些数字全部归零重新累计，你没法拿它当 checkpoint 用。

远端 reindex 还多一层坑：它要求目标集群在elasticsearch.yml里配置reindex.remote.whitelist，而且这个配置是静态的、改完要重启节点。如果迁移中途需要调整源端地址，等于要把整个集群滚一遍重启，这在生产上是很难接受的。

2.2 logstash-input-elasticsearch 的游标语义

Logstash 的 ES input 插件是另一条常见路线。它有三种docinfo模式，配合scroll使用，看起来像是能续传的——因为它确实提供了一个last_run_metadata_path参数，把上次运行的位置写到一个文件里。

但这里有个特别容易误解的点：这个元数据文件记录的是"上次 scroll 到的位置"，而不是可以精确续跑的文档游标。它的机制是在每次scroll返回时把当前 scroll id 和查询参数写进文件，下次启动时尝试用这个 scroll id 继续。问题在于，scroll 上下文是有生命周期的，默认 1 分钟，你就算设成scroll => "60m"，一旦进程挂掉超过一小时，这个 scroll id 就彻底失效了，重启后会回退到从头开始。

而且它记录的粒度是"当前 scroll 批次"，一个批次里可能有几千条文档，重启时这一批是整体重放还是整体跳过，取决于你的_id策略。如果用的是自动生成的_id，重放会产生重复文档（虽然 ES 会用新的_id再写一份，变成两份）；如果用的是稳定_id，重放是幂等的，只是白跑一批。

所以 Logstash 这条路适合什么场景？适合源端数据量中等、能接受偶发重复、且迁移期间可以随时重启进程的场景。它不适合亿级、不能重复、不能容忍整批重放的场景。

2.3 snapshot/restore 为什么天生不需要续传

快照这条路值的原理值得理解一下，因为它解释了"什么叫天然幂等"。快照是把 Lucene 段文件直接复制到仓库，恢复时把这些段文件挂载成索引。整个过程是文件级别的字节复制，中间断了之后重新发起，它只会传输那些还没传完的段文件。

这就带来两个特性：一是天然幂等，重复传同一个段文件不会产生副作用；二是天然增量，第二次快照只会传新增的段。所以"续传"这个概念在快照方案里根本不存在，因为它压根就没有"进度"这个状态需要维护。

代价是它的适用范围窄：两端版本必须兼容（跨大版本基本没戏）、仓库必须两边都能访问、索引不能做字段级转换。如果你的需求里包含"顺便把某个字段的类型改一下"或者"把两个索引合并成一个"，快照方案直接出局。

2.4 自研 scroll + bulk 方案的可控点在哪

排除掉上面三条之后，剩下的就是自己写脚本：从源端读，往目标端写，中间自己维护状态。这条路听起来最麻烦，但它的优势在于每一个环节你都能控制：

• 游标怎么定义，你说了算（scroll id、search_after 值、时间戳、自增 id 都行）
• checkpoint 存在哪，你说了算（本地文件、Redis、MySQL、甚至目标集群的一个小索引）
• 中断信号怎么处理，你说了算（捕获 SIGTERM，把当前批次刷完再退出）
• 幂等靠什么保证，你说了算（强制指定_id、用version_type=external、或者干脆先删后写）

代价当然也很明显：你需要自己处理 scroll 上下文过期、自己处理 bulk 部分失败、自己做限流、自己做监控埋点。下面两节就围绕这几件事展开。

3. 断点该存在哪：checkpoint 结构、游标选型与 slice 映射

这一节是整篇的核心。断点续传做得好不好，八成取决于 checkpoint 的设计是否一次到位，剩下的两成才是写入逻辑。

3.1 checkpoint 里到底要存哪几个字段

先明确一个原则：checkpoint 必须能回答"下一次从哪读"，而不能只回答"已经读了多远"。前者是游标，后者只是计数，计数是没法用来续跑的。

我在实际项目里用的结构大概长这样：

{ "job_id": "order_v3_to_cluster_b", "source_index": "order_v3", "target_index": "order_v3_new", "started_at": "2024-11-02T02:13:44+08:00", "slices": { "0": {"last_sort": [1730000000000, "doc_88213"], "done": false, "written": 12480000}, "1": {"last_sort": [1730000000000, "doc_88190"], "done": true, "written": 12500310}, "2": {"last_sort": null, "done": false, "written": 0} }, "total_written": 24980310, "pit_id": "gAEl...", "pit_created_at": "2024-11-02T02:13:50+08:00" }

几个字段的设计意图值得单独说：

last_sort是整个结构的灵魂，它存的不是文档 id，而是上次查询返回的完整排序值数组，也就是search_after需要的东西。为什么一定要存完整数组？因为如果排序用了多个字段，只存最后一个字段的值是没法定位的，search_after要求你把所有排序字段的值按顺序传回去。

done标记必须和last_sort分开存。有些切片的最后一批恰好返回空结果，此时last_sort还是上一次的值，如果你只用"返回空就停"，重启时会误判这个切片没跑完。

pit_id是可选项，但强烈建议存。PIT（point in time）能保证整个扫描过程看到的是同一个数据视图，避免迁移期间源端新增文档导致的翻页重复或遗漏。不过 PIT 也有寿命，默认 1 天，超过就失效，所以要在启动时先检查一下pit_created_at，太旧就重新开一个（重新开会丢失一致性保证，但至少能继续跑）。

job_id这个字段看着多余，其实是防呆用的。我曾经因为脚本里写死了 checkpoint 路径，误把 A 索引的 checkpoint 拿来续跑 B 索引的迁移，结果 B 索引从中间开始读，前半段全丢。加上job_id和source_index的校验，启动时对不上就直接报错，比事后查数据缺失省事一万倍。

3.2 用 search_after 做游标比 scroll_id 更适合续传

这是选型上最关键的一个决定，我把它单独拎出来讲。

scroll的机制是服务端维护一个上下文，客户端每次带上 scroll id 去取下一批。这个上下文存在服务端内存里，有 TTL，且无法序列化到磁盘。你把这个 scroll id 存到 checkpoint 里，进程重启后拿它去请求，大概率会收到search_context_missing_exception，因为节点可能已经重启过，或者上下文因为超时被清理了。

search_after完全不一样。它是个无状态的分页方式：你把上一批最后一条文档的排序值传回去，服务端就返回排序值大于它的那批文档。这个排序值是可以序列化、可以持久化、可以跨进程跨机器传递的，只要数据和排序规则不变，任何时候拿它去请求都能拿到正确结果。

代价是search_after有两个硬性要求：

第一，必须有确定的排序规则，而且这个规则要能唯一确定顺序。如果只按create_time排序，同一毫秒有多条文档时顺序是不稳定的，翻页可能重复或遗漏。所以必须加一个 tiebreaker，常用的是_id或者_shard_doc。

第二，排序字段不能为 null。这是最容易踩的坑。如果排序字段有缺失值，ES 默认把缺失值放在最后（_last），search_after传回一个正常值时就永远跳过不了那批缺失值，最后你会发现数量对不上，一查少了十几万条没有该字段的文档。解决方法是显式指定missing: "_last"并在脚本里单独处理最后一段，或者干脆换一个一定有值的字段做排序。

{ "size": 5000, "sort": [ {"create_time": {"order": "asc", "missing": "_last"}}, {"_shard_doc": "asc"} ], "pit": {"id": "gAEl...", "keep_alive": "30m"}, "search_after": [1730000000000, 88213] }

这里的_shard_doc是 PIT 提供的隐式分片内文档序，用它做 tiebreaker 比_id好，因为_id排序会触发 fielddata 加载，内存开销大；而_shard_doc是 Lucene 内部顺序，零成本。

如果目标集群版本不支持 PIT（比如源端是 6.x），退而求其次的做法是用_doc排序配合 scroll，或者找一个业务上一定存在的自增主键字段。用_doc的问题是它不保证跨查询稳定，段合并后顺序会变，所以只适合"源索引在迁移期间完全静止"的场景。

3.3 slice 切分与 checkpoint 的对应关系

单线程跑亿级索引太慢，所以一定要切片。切片的方案有几种：

• _reindex内置的slices: auto，但它的切片状态不对外暴露
• scroll 的slice: {id: 0, max: 8}，每个切片独立 scroll
• 手动按_id哈希或按时间范围切分成多个查询，每个查询有自己的 checkpoint

自研脚本里我推荐手动切片 + 每个切片独立 checkpoint，因为这样你能精确知道哪个切片卡住了，也能单独重启某个切片。代价是脚本要处理并发，得用线程池或者多进程。

切片数和 checkpoint 的对应关系必须是固定映射，不能动态调整。因为 checkpoint 里slices的 key 是切片 id，如果你第二次启动把max从 8 改成 16，那原来 0-7 号的 checkpoint 就没法对应了，只能全部作废重跑。所以切片数要在第一次运行时就想清楚，写进 checkpoint 里作为元数据。

切片数怎么定？我的经验是按目标集群的数据节点数 × 1.5 到 2 倍来定，同时不超过源索引的主分片数（超过也没用，一个分片同时只能被一个查询高效扫描）。比如源索引 12 个主分片、目标集群 6 个数据节点，那切 12 个切片是比较合理的。

还有一个细节：每个切片的search_after游标是独立维护的，不能混用。我曾经为了省事把所有切片的游标塞进一个列表里按顺序推进，结果切片的查询边界是重叠的，导致部分文档被两个切片同时读到，重复写入了好几百万条。虽然因为_id相同没造成数据错误，但白白浪费了两小时。

3.4 幂等写入：_id策略决定了重跑安不安全

断点续传能成立的前提是：同一批数据被写两次，结果和写一次一样。这就完全取决于_id怎么定。

三种情况：

第三种情况在小规模下不容易被发现，因为跑成功了看起来一切正常，等做数据校验的时候才发现总量比源端多了几十万。所以我的习惯是：迁移脚本里绝对禁止让目标端自动生成_id，永远从源文档的_id或者某个业务唯一键取值。

如果业务唯一键不唯一（比如同一个订单号有多条明细），那就用组合键，比如order_id + "_" + line_no拼成_id。千万不要用UUID或者时间戳生成，那样等于放弃了幂等性。

还有一个更严格的做法是用version_type=external配合版本号写入。它要求你给每个文档指定一个版本号（通常是源端的_version或者一个时间戳），目标端只有在版本号更高时才接受写入。好处是能防止并发写入导致的数据回退，坏处是版本号管理会更复杂，一般场景下用不上。

4. 一份能落地的续传脚本骨架，以及每个参数背后的取舍

原理讲够了，接下来是能跑的东西。下面这份骨架是基于 PIT + search_after + slice 的方式，用 Python 写的，语言选 Python 纯粹是因为改起来快，换成 Java 或 Go 逻辑完全一样。

4.1 索引准备：refresh_interval、副本、translog 的取舍

写代码之前先做目标索引的准备工作，这几项如果顺序搞错，性能会差好几倍。

# 1. 预建索引，显式指定 mapping 和 settings curl -XPUT "http://target:9200/order_v3_new" -H 'Content-Type: application/json' -d ' { "settings": { "number_of_shards": 12, "number_of_replicas": 0, "refresh_interval": "-1", "index.translog.durability": "async", "index.translog.sync_interval": "30s" } }'

几个参数的取舍理由：

refresh_interval: -1关掉自动刷新。默认每秒刷新一次会不断生成新的小段，写入过程中段数量爆炸，段合并压力巨大。迁移期间关掉它，写完再手动 refresh 一次，能省 30% 以上的时间。

number_of_replicas: 0关掉副本。写入时要同步到副本分片，等于双倍写入开销。迁移期间没有高可用需求，写完再设回去。

translog.durability: async是有数据丢失风险的，它把每次写操作从"同步刷盘"改成"每 30 秒刷一次"。如果节点在这 30 秒内崩溃，最近的数据会丢。所以这个参数我一般只在"能接受重跑"的场景下用，如果迁移数据不能丢，保持默认的request。

注意：这三项设置必须在写入之前改，写入过程中改虽然也能生效，但已经产生的小段和副本同步的开销是收不回来的。

预建索引还有一个更重要的原因：防止动态 mapping 把字段类型推断错。ES 默认会按第一条文档的字段值推断类型，如果第一条文档的某个字段恰好是"123"，它就被建成text而不是long，后面写入真正的数字会直接报错。所以我一般在预建 mapping 时加"dynamic": "strict"测一遍，确认所有字段都有定义，再改成false正式跑。

4.2 bulk 批次大小与requests_per_second的压测标定

批次大小不是一个拍脑袋的数字，它取决于文档平均大小和目标集群的写入能力。

BATCH_SIZE = 3000 # 条数 MAX_BULK_BYTES = 5 * 1024 * 1024 # 5MB，防止单批过大

为什么要两个限制？因为按条数限制在大文档场景下会失效。如果文档平均 20KB，3000 条就是 60MB，一次 bulk 请求直接把协调节点的内存打爆。所以必须再加一个字节数上限，两者取先到者。

一个粗略的标定方法：目标集群每 GB 堆内存对应约 5-8MB 的单批上限。如果目标数据节点是 16GB 堆、6 个节点，单批上限大概在 500MB 左右，看起来很大，但这是集群总承受能力，单客户端并发时不该用满。我的经验值是在 5-15MB 之间，通过压测找到那个"批量耗时稳定在 300-800ms"的点。

requests_per_second是限流参数，自研脚本里没有现成的，得自己实现令牌桶或者最简单的time.sleep()：

def rate_limit(rate_per_sec, batch_count): expected = batch_count / rate_per_sec actual = time.time() - batch_start if actual < expected: time.sleep(expected - actual)

限流的目的不是为了慢，而是为了别把源集群或目标集群打挂。迁移期间如果源集群还在承载线上查询，不加限流很容易把查询延迟打到秒级。我一般从目标集群实际吞吐 × 0.7开始试，观察两个集群的监控曲线，如果 CPU 稳定在 60% 以下、查询 P99 没有明显上升，再往上加。

4.3 主循环：读-存 checkpoint-写-更新游标

核心循环的顺序特别重要，我把它写成伪代码形式，因为顺序错了会导致两种完全不同的故障。

def run_slice(slice_id, ckpt): state = ckpt["slices"][str(slice_id)] if state["done"]: return search_after = state["last_sort"] while True: # 1. 先读一批 resp = es.search( index=SOURCE_INDEX, body=build_query(slice_id, SLICE_MAX, search_after), size=BATCH_SIZE, ) hits = resp["hits"]["hits"] if not hits: state["done"] = True save_ckpt(ckpt) # 空批次也要落盘 break # 2. 转成 bulk 动作 actions = [{"_index": TARGET_INDEX, "_id": h["_id"], "_source": h["_source"]} for h in hits] # 3. 写入，处理部分失败 ok, errors = bulk_with_retry(actions) # 4. 写成功之后才推进游标 state["last_sort"] = hits[-1]["sort"] state["written"] += ok save_ckpt(ckpt) # 5. 处理完再睡，别在写入前睡 rate_limit(RATE, len(actions))

这个顺序里有两个关键点：

游标必须在写入成功之后才推进。如果先推进游标再写入，一旦写入失败，重启后这批数据就被永久跳过了，而且你永远不会知道少了什么。反过来，先写入再推进游标，最坏情况是重启后重复写一批，因为有稳定_id所以幂等，只是浪费一点时间。宁可重复，不可遗漏，这是迁移的基本原则。

checkpoint 落盘的频率要权衡。每一批都落盘是最安全的，但每批都写一次磁盘（尤其是写到数据库）会拖慢整体速度。我的折中做法是：本地文件每批都写（开销极小），每 10 批同步一次到远端存储。这样最坏情况丢失 10 批的进度，重跑大概多花几十秒。

bulk_with_retry这里也有讲究。bulk 返回的errors: true不代表整批都失败，它是部分失败，items数组里会标明哪几条失败、失败原因是什么。常见原因有两类：一类是可重试的（es_rejected_execution_exception，队列满了），一类是不可重试的（mapper_parsing_exception，字段类型不对）。前者退避重试，后者必须记下来人工处理，否则会陷入无限重试。

def bulk_with_retry(actions, max_retry=3): for attempt in range(max_retry): resp = es.bulk(operations=flatten(actions), refresh=False) if not resp["errors"]: return len(actions), [] retry_actions, fatal = split_errors(resp, actions) if not retry_actions: return len(actions) - len(fatal), fatal actions = retry_actions time.sleep(2 ** attempt) return 0, fatal

4.4 中断信号处理与优雅退出

这一段是很多人会漏掉的。脚本收到Ctrl+C或者容器收到SIGTERM时，默认行为是立刻退出，而此时可能刚好有一批 bulk 请求在飞。这批数据到底写没写进去？不知道。下次启动就会重复写或者漏写。

正确做法是注册信号处理器，让主循环感知到"该停了"，把当前这批做完、checkpoint 刷盘、再退出：

STOP = False def handle_signal(signum, frame): global STOP STOP = True signal.signal(signal.SIGTERM, handle_signal) signal.signal(signal.SIGINT, handle_signal) # 主循环里 while not STOP: ... # 退出前 save_ckpt(ckpt, flush=True)

这里有个细节值得注意：信号处理器里只设置标志位，不做任何 IO。因为在信号处理函数里调用写文件、发网络请求都是不安全的，可能引发死锁。把实际工作留给主循环，这是标准做法。

容器环境还有一个额外问题：K8s 发送SIGTERM之后默认只等 30 秒就SIGKILL。如果你的批量耗时可能有 10 秒，那要保证在收到信号后立刻停止发新请求、等当前批完成、然后快速刷盘退出，整个流程控制在 20 秒以内比较安全。可以通过terminationGracePeriodSeconds把它调到 60 秒，给自己留余量。

5. 恢复之后先别急着切别名：校验、补齐与收尾顺序

脚本跑完打印"全部完成"的那一刻，是最容易出错的时候。很多人直接切别名上线，然后过两天发现少了 20 万条数据。校验这一步不能省，而且顺序有讲究。

5.1 数量对不上时的三种排查路径

第一步永远是_count对比：

curl "http://source:9200/order_v3/_count" curl "http://target:9200/order_v3_new/_count"

如果数量一致，不代表数据一致（可能有重复+遗漏正好抵消），但至少说明大概率没问题。如果数量不一致，按下面三条路径排查：

路径一：是不是切片边界重叠导致的重复。查一下 checkpoint 里各切片的written总和，和最终_count对比。如果总和大于_count，说明有重复，但因为_id相同被覆盖了，实际数据可能是对的。如果总和小，说明有遗漏。

路径二：是不是排序字段有空值。这是最常见的遗漏原因。写一条查询{"exists": {"field": "create_time"}}数一下有多少条缺少排序字段，如果这个数字刚好等于差额，那就对上了。处理方法很简单：单独补跑一个不带search_after的查询，把所有缺失该字段的文档拉出来补写一遍。

路径三：是不是迁移期间源端有删除。迁移是扫快照，如果扫描开始后源端删了文档，目标端会多出来。这种"多"是正常的，需要跟业务确认是否要把这些删除同步过去。

还有一种更隐蔽的情况：目标端的_count包含了尚未 refresh 的文档。ES 的_count会等到 refresh 之后才准确，如果你刚写完就查，可能查到的是旧值。所以校验前一定要先手动 refresh：

curl -XPOST "http://target:9200/order_v3_new/_refresh"

5.2 版本冲突、mapping 冲突、字段丢值的处理

跑完一轮之后，把 bulk 的失败日志捞出来分类统计，这是最有价值的复盘材料。我遇到过的失败大致分这几类：

mapper_parsing_exception这类错误有个特点：它是永久性的，重试多少次都不会成功。所以脚本里必须把它单独捞出来记到一个 dead letter 文件里，跑完之后人工处理，不能让整个任务卡在这里。

字段丢值这个问题更隐蔽，因为它不报错。常见原因有两个：一是目标 mapping 里某个字段设了"index": false或者"enabled": false，值写进去了但查不出来；二是日期格式不匹配，strict_date_optional_time和epoch_millis之间转换失败时，如果字段设了ignore_malformed: true，值会被静默丢弃。

排查方法是对某个字段做一个 terms 聚合，两边对比 top N 的分布：

{"size": 0, "aggs": {"status": {"terms": {"field": "status", "size": 20}}}}

两边跑一遍，如果分布明显不同，说明这个字段有问题。

5.3 别名切换与副本恢复的顺序

收尾操作的顺序我踩过坑，所以固定成下面这个流程：

1. 先 refresh，让所有数据对查询可见
2. 跑校验，数量 + 抽样 + 聚合对比
3. 恢复副本数，number_of_replicas设回正常值
4. 恢复 refresh_interval，设回1s或业务需要的值
5. 等副本追平，查_cat/shards确认所有副本都是STARTED且没有INITIALIZING
6. 切别名，用_aliases原子操作
7. 观察一段时间再下线旧索引

第 5 步特别重要。如果副本还没追平就切别名，切过去的瞬间所有查询压力会全压在主分片上，很容易把集群打挂。查副本状态：

curl "http://target:9200/_cat/shards/order_v3_new?v&h=index,shard,prirep,state,docs"

切别名用原子操作，一步到位，不要先删后加：

{ "actions": [ {"remove": {"index": "order_v3", "alias": "order"}}, {"add": {"index": "order_v3_new", "alias": "order"}} ] }

还有一个容易被忽略的点：forcemerge别在切别名之后做。段合并会消耗大量 IO，如果这时候线上流量已经切过来了，查询延迟会明显抖动。要合并就在切之前做，或者干脆不做（ES 自己的后台合并最终会收敛，只是慢一点）。如果确实要做，forcemerge的max_num_segments设成 1 只对只读索引有意义，对还在写入的索引反而有害。

6. 几处踩过才知道的细节

最后分享几个文档里不会写、但会实实在在影响迁移结果的经验。

PIT 的 keep_alive 要设得比单批耗时大一个数量级。默认 1 分钟，如果你的批量处理加上重试可能超过 1 分钟，PIT 就会在翻页过程中过期，报search_context_missing_exception。我一般设成30m，同时在脚本里捕获这个异常，重新开 PIT 并从最后一批的last_sort继续——因为last_sort是数据层面的游标，跟 PIT 无关，所以恢复是安全的，只是会丢失"一致性视图"这个保证。

迁移期间源端的 refresh 频率会影响你看到的数据量。如果源索引的refresh_interval是 30 秒，那你扫到的数据是"最多 30 秒前的"，迁移结束前源端最后写的那批可能不在你扫到的范围里。所以全量跑完之后必须再做一次增量追平，用create_time大于等于迁移开始时间做条件，补一轮。

不要用_id做排序字段。前面提过一次，这里再强调：_id排序会加载 fielddata，一个亿级索引可能吃掉几十 GB 堆内存，直接把节点拖垮。要么用_shard_doc（有 PIT 的情况），要么用业务字段加唯一性 tiebreaker。

判断"完成"不能只看返回空。有时候查询返回空是因为超时被降级，或者 PIT 过期，而不是真的读完了。脚本里要做区分：返回hits.total.value为 0 且没有异常才算完成；如果是异常导致的空，要重试而不是标记done。这个区别在处理亿级索引时特别重要，因为偶发的 shard 失败很常见。

迁移日志里一定要打文档的_id。出问题时能精确定位到哪条文档导致了失败，比"第 12483 批有 3 条失败"有用得多。日志量大的话可以只打失败的_id，成功的用一个计数器就够。

最后一个小技巧：如果你要在多个环境重复做这类迁移，把 checkpoint 的存储抽象成接口（本地文件、Redis、MySQL 三种实现），脚本主体完全不动。我后来把整套逻辑封装成了一个带 CLI 的小工具，--resume参数一加就能续跑，--dry-run只读不写，--verify只对比不迁移，实际用下来比每次改脚本省事太多。这套东西的价值不在于代码多优雅，而在于当你半夜被叫起来处理一个"迁移任务挂了"的问题时，你只需要敲一条命令，剩下的它自己会接着干。