Elasticsearch `_reindex` 中断了怎么办？详解断点续传与重启策略

`_reindex` 的“脆弱”时刻：为何中断如此棘手？

当你启动一个庞大的 Elasticsearch _reindex 任务，比如需要迁移数十亿文档、调整 mapping 或进行版本升级时，最担心的事情莫过于任务中途意外中断。节点故障、网络抖动、OOM（Out of Memory）错误，甚至只是手误取消了任务，都可能让数小时甚至数天的进度付诸东流。

为什么 _reindex 的中断恢复不像我们想象的那么简单直接呢？核心原因在于：标准的 _reindex API 本身，在设计上并不原生支持精细化的断点续传。

当你发起一个 POST /_reindex 请求时，Elasticsearch 在后台实际上做了几件事：

启动一个任务（Task）：你可以通过 Task Management API (_tasks) 查到这个任务。
执行滚动查询（Scroll Query）：从源索引 (source.index) 分批次地拉取文档数据。
执行批量写入（Bulk Indexing）：将拉取到的文档写入目标索引 (dest.index)。

关键在于，对于一次性的 _reindex 调用（即使设置了 wait_for_completion=false 让它在后台运行），Elasticsearch 并不会 持久化记录“我已经处理到源索引的哪个具体位置”或者“哪些文档已经成功写入目标索引”。任务信息（Task Info）里会显示已处理的文档数、是否有版本冲突等统计信息，但这更像是运行时的计数器，而不是一个可以在任务失败重启后用于恢复进度的状态标记。

一旦任务因为节点崩溃、网络中断或被取消而彻底终止（不是暂时性的节流 throttled），这个运行时的上下文就丢失了。下次你重新发起同样的 _reindex 请求，它通常会从头开始，尝试重新读取源索引的所有文档，并写入目标索引。这不仅浪费了之前完成的工作，还可能导致目标索引出现大量重复文档（如果目标索引没有使用源文档 ID 作为其 _id，或者允许重复 ID）。

因此，指望简单地重新运行 _reindex 命令就能自动从上次中断的地方继续，是不现实的。我们需要更“聪明”的策略。

策略一：利用 `query` 参数实现手动分片与续传

这是最常用也是相对容易理解的“模拟”断点续传的方法。核心思想是：不一次性处理整个源索引，而是通过在 _reindex 请求中加入 query 参数，将大任务分解成一系列只处理源索引部分数据的小任务。 当某个小任务中断后，你只需要重新运行该特定范围的任务，或者从下一个范围开始。

如何定义“范围”？

通常依赖于源索引中存在某种可以排序和划分范围的字段。常见的选择包括：

时间戳字段（@timestamp 或类似字段）：如果你的数据有明确的时间序列特征，这是最自然的选择。
数字类型的 ID 或序列号字段：如果文档有自增 ID 或其他数值型标识符。
唯一标识符的哈希或前缀：虽然不太常用，但理论上也可以。

具体操作步骤：

假设我们有一个源索引 my_source_index，目标索引 my_target_index，并且源索引有一个 @timestamp 字段。

确定划分逻辑：决定按什么时间间隔（天、小时、分钟）或数量级（ID 范围）来划分任务。例如，我们决定按天来处理。
编写带 query 的 _reindex 请求：

POST _reindex?requests_per_second=500&wait_for_completion=false
{
  "source": {
    "index": "my_source_index",
    "query": {
      "range": {
        "@timestamp": {
          "gte": "2023-10-01T00:00:00Z",
          "lt": "2023-10-02T00:00:00Z" 
        }
      }
    },
    "size": 1000 // 每次滚动查询拉取的文档数，可以根据情况调整
  },
  "dest": {
    "index": "my_target_index",
    "op_type": "create" // 重要：如果希望避免重复写入（假设ID相同），使用 create
  }
}

query：这里使用了 range 查询来限定只处理 @timestamp 在 2023年10月1日这一天的数据。
gte (greater than or equal to) 和 lt (less than)：定义了时间范围。注意区间的开闭，确保不遗漏、不重复。
op_type: "create"：这是一个重要的选项。如果目标索引的文档 ID (_id) 与源索引相同，create 操作会确保只有当目标索引中不存在该 ID 时，文档才会被写入。这能有效防止因任务重试导致的重复文档。如果你的目标索引允许重复 ID 或者有不同的 ID 策略，则可能不需要或使用默认的 index 操作。
requests_per_second：控制 reindex 的速率，防止对集群造成过大压力。设置为 -1 表示不限制（不推荐用于大索引）。
wait_for_completion=false：让 reindex 请求立即返回一个 task ID，任务在后台异步执行。你需要记下这个 task ID。

监控任务：使用 Task Management API 监控每个小任务的进度。

GET _tasks/<task_id>

关注返回结果中的 completed 状态以及 response.total, response.created, response.updated, response.deleted, response.failures 等字段。

处理中断与续传：

如果一个小任务（例如处理 10月1日数据的任务）成功完成：记录下来，然后提交处理下一个范围（例如 10月2日）的任务。
如果一个小任务中途失败或被取消：
* 首先，分析失败原因（检查 Elasticsearch 日志、监控指标等）。
* 修复问题（例如，增加节点资源、解决网络问题）。
* 重新提交处理相同范围（10月1日）的任务。由于我们使用了 op_type: "create"（假设 ID 策略允许），即使部分数据之前已经成功写入，重新运行时这些文档会被跳过（产生 version conflict，但不会重复写入），只会处理之前未成功的文档。如果未使用 op_type: "create" 或 ID 不同，则可能需要更复杂的逻辑，比如先删除目标索引中对应范围的数据再重跑，但这通常更麻烦且有风险。

自动化脚本：对于大量数据，手动管理这些小任务非常繁琐。最佳实践是编写一个外部脚本（Python, Bash 等）来自动执行以下操作：
- 计算需要处理的所有范围（例如，根据源索引的最小和最大时间戳生成每日范围列表）。
- 维护一个状态记录（例如，一个简单的文件或数据库），标记哪些范围已经成功完成。
- 循环遍历未完成的范围，提交 _reindex 任务。
- 定期检查任务状态，更新状态记录。
- 处理失败的任务（记录日志、尝试重试、或者发出警报）。

优点：

概念简单：利用了 _reindex API 已有的 query 功能。
相对可控：将大问题分解为小问题，降低了单次失败的影响范围。
不依赖外部工具：纯粹使用 Elasticsearch API 即可实现（如果配合脚本管理）。

缺点：

需要手动管理（或脚本开发）：需要自行实现范围划分、任务提交、状态跟踪和重试逻辑。
范围定义可能不完美：如果依赖的字段（如 @timestamp）分布不均，可能导致某些范围的任务量远超其他范围。
op_type: "create" 的依赖：防止重复写入的效果很大程度上依赖于源和目标索引使用相同的 _id，并且你使用了 create 操作。如果场景不符，处理重复数据会更复杂。
对源索引的查询压力：每个小任务都需要对源索引执行一次 range 查询。

策略二：利用 Sliced Scrolling 提高并行度（但不能直接解决断点续传）

Sliced Scrolling 是 _reindex 和 _update_by_query 等 API 支持的一个特性，允许将滚动查询（Scroll）在内部划分为多个“切片”（Slices），并行地从源索引读取数据。这可以显著提高 _reindex 的速度，尤其是在源索引有多个分片（Shard）的情况下。

POST _reindex?requests_per_second=500&wait_for_completion=false
{
  "source": {
    "index": "my_source_index",
    "slice": {
      "id": 0, // 当前切片的 ID (从 0 开始)
      "max": 5  // 总共划分的切片数量
    }
  },
  "dest": {
    "index": "my_target_index"
  }
}

你需要为每个切片（id 从 0 到 max-1）分别提交一个 _reindex 请求。Elasticsearch 会根据文档的 _id 或路由（如果自定义了）将文档分配到不同的切片进行处理。

注意： Sliced Scrolling 本身并不直接解决断点续传的问题。它主要是为了并行化读取操作以提高效率。如果你提交了 5 个切片的 _reindex 任务，其中一个切片对应的任务失败了，你需要重新运行该失败切片对应的任务。它并没有记录这个切片内部处理到了哪个文档。

如何结合使用？

你可以将策略一（按 query 划分范围）和 Sliced Scrolling 结合起来：

先按时间或 ID 范围划分成较大的任务块（例如，按天）。
对于每个任务块，再使用 Sliced Scrolling 将其内部处理并行化。

例如，处理 10月1日的数据时，可以同时提交 5 个 _reindex 请求，每个请求处理 10月1日数据的一个切片：

// 请求 1 (切片 0)
POST _reindex?...
{ "source": { "index": "my_source_index", "query": { "range": { "@timestamp": { ... } } }, "slice": { "id": 0, "max": 5 } }, ... }

// 请求 2 (切片 1)
POST _reindex?...
{ "source": { "index": "my_source_index", "query": { "range": { "@timestamp": { ... } } }, "slice": { "id": 1, "max": 5 } }, ... }

// ... 以此类推直到切片 4

如果处理 10月1日数据的任务中，切片 2 失败了，你只需要重新提交处理 10月1日数据、切片 ID 为 2 的那个 _reindex 请求。

这种组合方式兼顾了范围控制（用于恢复）和并行处理（用于加速），但管理起来更加复杂，通常需要更健壮的自动化脚本。

策略三：利用外部工具实现更可靠的断点续传（如 Logstash）

当内置的 _reindex 及其变通方法在可靠性和管理复杂度上无法满足需求时，可以考虑使用外部工具来编排整个数据迁移过程。Logstash 是一个常见的选择，尤其因为它具有持久化队列（Persistent Queue） 的能力。

Logstash 如何工作？

一个典型的使用 Logstash 进行数据迁移（类似 reindex）的 pipeline 配置可能如下：

input {
  elasticsearch {
    hosts => ["your_source_es_host:9200"]
    index => "my_source_index"
    query => '{ "query": { "match_all": {} } }' // 或者更复杂的查询
    scroll => "5m"
    size => 1000
    docinfo => true
    # 可以添加 schedule 来定期运行，或一次性运行
  }
}

filter {
  # 可选：在这里进行数据转换、修改字段等操作
  # mutate { remove_field => ["@version"] }
}

output {
  elasticsearch {
    hosts => ["your_target_es_host:9200"]
    index => "my_target_index"
    document_id => "%{[@metadata][_id]}" // 使用源文档 ID 作为目标文档 ID
    action => "create" // 同样，使用 create 防止重复
  }
}

关键在于 Logstash 的持久化队列：

启用持久化队列：需要在 logstash.yml 中配置 queue.type: persisted。
工作原理：
- Input 插件（如 elasticsearch input）从源读取数据，并将事件（Events）放入队列。
- Output 插件（如 elasticsearch output）从队列中取出事件，尝试写入目标。
- 只有当 Output 插件确认事件已成功写入目标后，该事件才会被从队列中移除。
断点续传能力：
- 如果 Logstash 进程在处理过程中意外终止（例如，机器重启、进程崩溃），队列中尚未被成功处理（即未被 Output 确认）的事件会保留在磁盘上。
- 当 Logstash 重启后，它会读取磁盘上的持久化队列，自动从上次中断的地方继续处理，将队列中剩余的事件发送给 Output 插件。
- 结合 elasticsearch input 插件内部对 scroll 的管理（它会记录 scroll_id 并尝试恢复），以及 elasticsearch output 插件的 action => "create"，可以实现相当可靠的断点续传。

优点：

更强的容错性：Logstash 的持久化队列专门设计用于处理中断，状态管理更可靠。
解耦：将数据读取、转换（可选）和写入的逻辑分离到专门的工具中，主 Elasticsearch 集群只需响应读写请求。
灵活性：Logstash 提供了丰富的插件生态系统，可以在迁移过程中进行复杂的数据转换和处理。
内置重试机制：Output 插件通常内置了对目标写入失败的重试逻辑。

缺点：

引入额外组件：需要部署和管理 Logstash 实例。
配置和调优：需要学习 Logstash 的配置语法，并对其性能进行调优（如 worker 数量、batch size、JVM 设置等）。
性能可能不如原生 _reindex：数据需要经过 Logstash 中转，可能会引入额外的延迟和资源开销，尤其是在网络传输和序列化/反序列化方面。原生 _reindex 是在 Elasticsearch 集群内部节点间直接进行数据传输，通常效率更高。
对源 ES 的 Scroll API 依赖：Logstash input 插件仍然依赖源 ES 的 Scroll API 来拉取数据，如果 Scroll 上下文因长时间中断而过期，可能仍需要一些手动干预或配置调整（比如重新启动 Logstash pipeline，它会发起新的 Scroll 请求）。

监控与最佳实践

无论采用哪种策略，以下几点对于成功执行大规模 _reindex 并处理中断至关重要：

充分监控：
- 任务进度：使用 Task Management API (GET _tasks?actions=*reindex*&detailed=true) 持续跟踪任务状态、已处理文档数、版本冲突、失败等。
- 集群健康：监控 Elasticsearch 集群的 CPU、内存、磁盘 I/O、网络、JVM Heap 使用率、线程池（尤其是 write 和 search）队列和拒绝情况。确保集群能够承受 _reindex 带来的负载。
- 日志：密切关注源和目标集群的 Elasticsearch 日志，以及 Logstash（如果使用）的日志，查找错误和警告信息。
合理调优：
- requests_per_second：设置一个合理的速率限制，避免压垮集群。从小处开始，根据监控情况逐步增加。
- size (source batch size)：源查询每次拉取的文档数。影响内存使用和 scroll 效率。
- Bulk Size (implicit in _reindex, configurable in Logstash output)：目标索引每次批量写入的文档数。太小效率低，太大可能导致 OOM 或超时。
- Sliced Scrolling (slice)：根据源索引分片数和节点 CPU 核心数合理设置切片数量，通常等于源索引主分片数可以获得较好的并行效果。
- 目标索引设置：在 _reindex 开始前，优化目标索引的设置，例如暂时禁用 refresh_interval (index.refresh_interval: -1) 和减少副本数量 (index.number_of_replicas: 0)，待完成后再恢复。这可以显著提高写入性能。
测试先行：在生产环境执行大规模 _reindex 前，务必在测试环境中使用类似规模的数据（或按比例缩小）进行充分测试，验证所选策略的有效性、性能和恢复流程。
ID 策略与 op_type：仔细考虑源文档 ID 和目标文档 ID 的关系。如果希望保持唯一性并利用 op_type: "create" 防止重复，确保 ID 映射正确。如果目标索引需要生成新的 ID，那么处理重复会更复杂，可能需要在失败恢复时先清理目标索引中对应范围的数据。
资源规划：确保 Elasticsearch 集群（源和目标，如果分开的话）以及 Logstash（如果使用）有足够的 CPU、内存、磁盘空间和网络带宽来支持 _reindex 操作。

总结：没有银弹，只有权衡

处理 Elasticsearch _reindex 中断并没有一劳永逸的完美方案。你需要根据你的具体场景、数据量、对可靠性的要求、团队的技术栈以及可接受的复杂度来选择最合适的策略：

追求简单、纯 ES 方案：优先考虑策略一（query 参数分片），并结合自动化脚本进行管理。如果性能瓶颈在读取端，可以进一步结合策略二（Sliced Scrolling）。
追求高可靠性、容错性，且不介意引入外部工具：策略三（使用 Logstash 持久化队列） 是一个非常健壮的选择，尤其适合需要复杂转换或对中断恢复要求极高的场景。

无论选择哪种方式，周密的计划、充分的监控、合理的调优以及必要的自动化都是确保大型 _reindex 任务顺利进行或在中断后能有效恢复的关键。记住，与其在中断后手忙脚乱，不如在一开始就设计一个具有恢复能力的流程。