如何让Ingest Pipeline仅处理新增文档并输出至新索引？

核心思路 #

先完成存量文档的批量处理与迁移，再配置让后续新增文档自动触发Pipeline，同时避免重复处理存量数据。

具体步骤 #

1. 确认存量文档的迁移处理完成 #

先确保你执行的_reindex任务已全部完成，所有存量文档从index*迁移到new_index并通过Pipeline处理完毕。可通过以下命令查看任务状态：

GET _tasks?actions=*reindex&detailed=true

2. 为目标索引设置默认Pipeline #

将你的pipeline设为new_index的默认Ingest Pipeline，后续所有写入/更新到new_index的文档会自动触发该Pipeline处理，存量文档因已处理过，不会被重复触发（除非手动更新）。执行命令：

PUT /new_index/_settings
{
  "index.default_pipeline": "pipeline"
}

3. 停止全量_reindex的重复执行 #

如果你之前定时执行全量_reindex来关联Pipeline，必须停止该操作，否则会重复处理已完成的存量文档，造成资源浪费。

可选：避免存量文档更新时重复处理 #

若存量文档后续有更新需求，且不希望更新操作触发Pipeline处理，可修改Pipeline添加条件判断：

方案A：用标记字段判断是否已处理

给Pipeline添加条件处理器，仅未标记的文档会执行处理逻辑，处理后自动添加标记：

PUT _ingest/pipeline/pipeline
{
  "processors": [
    {
      "condition": "ctx._source?.processed != true",
      "gsub": {
        "field": "my_field",
        "pattern": "regex",
        "replacement": ""
      }
    },
    {
      "set": {
        "field": "processed",
        "value": true,
        "ignore_failure": true
      }
    }
  ]
}

方案B：用时间条件过滤

设置仅处理_reindex完成时间之后的文档，将示例时间替换为你的实际完成时间：

PUT _ingest/pipeline/pipeline
{
  "processors": [
    {
      "condition": "ctx._source['@timestamp'] > '2024-05-20T00:00:00Z'",
      "gsub": {
        "field": "my_field",
        "pattern": "regex",
        "replacement": ""
      }
    }
  ]
}

内容的提问来源于stack exchange，提问作者fehim