使用故障存储来解决摄取问题
当摄取过程中出现问题时，通常这不是一个孤立的事件。为了方便起见，本文提供了一些示例，展示了如何使用故障存储区来快速响应摄取失败并使索引恢复正常。
当文档在摄取管道中失败时，很难准确找出哪里出了问题以及出在什么地方。当这些失败在摄取过程的这个阶段被故障存储区捕获时，它们将包含额外的调试信息。失败的文档会记录处理器类型以及发生失败时正在执行的管道。失败的文档还将包含一个管道跟踪（pipeline trace），用于记录文档在失败时所处的任何嵌套管道调用。
为了演示这一点，我们将跟踪一个失败的文档穿过一个不熟悉的数据流和摄取管道的过程
POST my-datastream-ingest/_doc
{
"@timestamp": "2025-04-21T00:00:00Z",
"important": {
"info": "The rain in Spain falls mainly on the plain"
}
}
{
"_index": ".fs-my-datastream-ingest-2025.05.09-000001",
"_id": "F3S3s5YBwrYNjPmayMr9",
"_version": 1,
"result": "created",
"_shards": {
"total": 1,
"successful": 1,
"failed": 0
},
"_seq_no": 2,
"_primary_term": 1,
"failure_store": "used"
}
- 文档已被发送到故障存储区。
现在我们搜索故障存储区以检查失败的文档，看看哪里出了问题。
GET my-datastream-ingest::failures/_search
{
"took": 0,
"timed_out": false,
"_shards": {
"total": 1,
"successful": 1,
"skipped": 0,
"failed": 0
},
"hits": {
"total": {
"value": 1,
"relation": "eq"
},
"max_score": 1,
"hits": [
{
"_index": ".fs-my-datastream-ingest-2025.05.09-000001",
"_id": "F3S3s5YBwrYNjPmayMr9",
"_score": 1,
"_source": {
"@timestamp": "2025-05-09T06:24:48.381Z",
"document": {
"index": "my-datastream-ingest",
"source": {
"important": {
"info": "The rain in Spain falls mainly on the plain"
},
"@timestamp": "2025-04-21T00:00:00Z"
}
},
"error": {
"type": "illegal_argument_exception",
"message": "field [info] not present as part of path [important.info]",
"stack_trace": """j.l.IllegalArgumentException: field [info] not present as part of path [important.info]
at o.e.i.IngestDocument.getFieldValue(IngestDocument.java:202)
at o.e.i.c.SetProcessor.execute(SetProcessor.java:86)
... 19 more
""",
"pipeline_trace": [
"ingest-step-1",
"ingest-step-2"
],
"pipeline": "ingest-step-2",
"processor_type": "set"
}
}
}
]
}
}
- 当摄取管道失败时，存储的文档就是最初发送到集群的内容。
- 我们未能找到的重要信息最初是存在于文档中的。
- 发生故障时，
info字段不存在。 - 第一个管道调用了第二个管道。
- 该文档在第二个管道中失败。
- 它在管道的
set处理器中失败了。
尽管事先不了解这些管道，但我们有一些可以开始排查的地方。尽管 important.info 字段存在于发送到集群的文档中，但 ingest-step-2 管道却找不到它。如果我们拉取该管道的定义，会发现以下内容
GET _ingest/pipeline/ingest-step-2
{
"ingest-step-2": {
"processors": [
{
"set": {
"field": "copy.info",
"copy_from": "important.info" <2>
}
}
]
}
}
- 这里只有一个处理器。
- 此时文档中缺少该字段。
ingest-step-2 管道中只有一个 set 处理器，因此这很可能不是根本问题所在。回想一下故障记录中的 pipeline_trace 字段，我们发现 ingest-step-1 是为此文档调用的原始管道。它很可能是数据流的默认管道。拉取其定义，我们发现以下内容
GET _ingest/pipeline/ingest-step-1
{
"ingest-step-1": {
"processors": [
{
"remove": {
"field": "important.info"
}
},
{
"pipeline": {
"name": "ingest-step-2"
}
}
]
}
}
- 一个错误地移除了我们重要字段的
remove处理器。 - 对第二个管道的调用。
我们在第一个管道中发现了一个 remove 处理器，这就是问题的根本原因！应更新管道以不移除重要数据，或者更改下游管道，使其不再期望重要数据总是存在。
摄取处理器可以打上标签。这些标签是用户提供的信息，用于命名或描述处理器在管道中的用途。当文档由于处理器问题被重定向到故障存储区时，它们会捕获发生故障的处理器中的标签（如果存在）。由于这种行为，为管道中的处理器打标签是一个好习惯，这样可以快速定位故障位置。
这里我们有一个不必要地复杂的管道。它由几个 set 和 remove 处理器组成。幸运的是，它们都打了描述性的标签。
PUT _ingest/pipeline/complicated-processor
{
"processors": [
{
"set": {
"tag": "initialize counter",
"field": "counter",
"value": "1"
}
},
{
"set": {
"tag": "copy counter to new",
"field": "new_counter",
"copy_from": "counter"
}
},
{
"remove": {
"tag": "remove old counter",
"field": "counter"
}
},
{
"set": {
"tag": "transfer counter back",
"field": "counter",
"copy_from": "new_counter"
}
},
{
"remove": {
"tag": "remove counter again",
"field": "counter"
}
},
{
"set": {
"tag": "copy to new counter again",
"field": "new_counter",
"copy_from": "counter"
}
}
]
}
我们摄取了一些数据，发现它被发送到了故障存储区。
POST my-datastream-ingest/_doc?pipeline=complicated-processor
{
"@timestamp": "2025-04-21T00:00:00Z",
"counter_name": "test"
}
{
"_index": ".fs-my-datastream-ingest-2025.05.09-000001",
"_id": "HnTJs5YBwrYNjPmaFcri",
"_version": 1,
"result": "created",
"_shards": {
"total": 1,
"successful": 1,
"failed": 0
},
"_seq_no": 1,
"_primary_term": 1,
"failure_store": "used"
}
在检查故障时，我们可以快速识别出导致问题的带标签处理器。
GET my-datastream-ingest::failures/_search
{
"took": 0,
"timed_out": false,
"_shards": {
"total": 1,
"successful": 1,
"skipped": 0,
"failed": 0
},
"hits": {
"total": {
"value": 1,
"relation": "eq"
},
"max_score": 1,
"hits": [
{
"_index": ".fs-my-datastream-ingest-2025.05.09-000001",
"_id": "HnTJs5YBwrYNjPmaFcri",
"_score": 1,
"_source": {
"@timestamp": "2025-05-09T06:41:24.775Z",
"document": {
"index": "my-datastream-ingest",
"source": {
"@timestamp": "2025-04-21T00:00:00Z",
"counter_name": "test"
}
},
"error": {
"type": "illegal_argument_exception",
"message": "field [counter] not present as part of path [counter]",
"stack_trace": """j.l.IllegalArgumentException: field [counter] not present as part of path [counter]
at o.e.i.IngestDocument.getFieldValue(IngestDocument.java:202)
at o.e.i.c.SetProcessor.execute(SetProcessor.java:86)
... 14 more
""",
"pipeline_trace": [
"complicated-processor"
],
"pipeline": "complicated-processor",
"processor_type": "set",
"processor_tag": "copy to new counter again"
}
}
}
]
}
}
- 这很有帮助，但会是管道上的哪个
set处理器呢？ - 文档在其上失败的确切处理器的标签。
如果没有设置标签，索引问题发生在管道的哪个位置就不会那么清晰。标签为处理器提供了唯一标识符，在发生摄取故障时可以快速引用。
由于可以像搜索普通数据流一样搜索故障存储区，我们可以将它们用作 Kibana 中警报规则的输入。以下是一个简单的警报示例，当数据流在过去五分钟内发生超过十次索引失败时触发该警报
-
创建故障存储区数据视图
如果你想使用 KQL 或 Lucene 查询类型，应首先为你的故障存储区数据创建一个数据视图。如果你计划使用 ES|QL 或 Query DSL 查询类型，则不需要此步骤。
导航到 Kibana 中的数据视图页面并添加一个新的数据视图。使用选择器语法将索引模式设置为你的故障存储区。
-
创建新规则
导航到 Management / Alerts and Insights / Rules。创建一个新规则。选择 Elasticsearch 查询选项。
-
选择你的查询类型
选择你希望使用的查询类型
对于 KQL/Lucene 查询，引用包含你的故障存储区的数据视图。
对于 Query DSL 查询，请在你的数据流名称后使用
::failures后缀。
对于 ES|QL 查询，请在
FROM命令的数据流名称后使用::failures后缀。
-
测试
在保存规则之前，配置警报的时间表、操作和详细信息。
-
完成
如果你遇到了长期的摄取失败，你可能会发现在你的数据流中出现了相当大的事件空白。如果启用了故障存储区，本应填补这些空白的文档将被隐藏在数据流的故障存储区中。由于故障存储区由常规索引组成，并且失败文档包含了失败的文档源，因此失败文档通常可以重放到你的生产数据流中。
警告
从故障存储区将数据重放到数据流中时应小心谨慎。重放过程中的任何失败都可能在故障存储区中产生新的失败，这可能会重复并掩盖原始事件。
我们针对修复故障数据推荐了一些最佳实践。
事先分离你的故障。 如前一节故障文档源所述，根据文档在摄取过程中失败的时间，故障文档的结构有所不同。我们建议至少按摄取管道故障和索引故障对文档进行分类。摄取管道故障通常需要重新运行原始管道，而索引故障则应跳过任何管道。按索引或特定故障类型进一步分离故障也可能会有所帮助。
执行故障存储区滚动更新。 在尝试修复故障之前，请考虑滚动更新故障存储区。这将创建一个新的故障索引，用于收集修复过程中的任何新故障。
使用摄取管道将故障文档转换回其原始文档。 故障文档存储了故障信息以及摄取失败的文档。修复文档的第一步应该是使用摄取管道从故障文档中提取原始源，然后丢弃关于故障的任何其他信息。
首先进行模拟以避免重复失败。 如果你必须在修复过程中运行管道，最好先针对该故障模拟管道。这将捕获任何可能导致文档第二次失败的不可预见的问题。请记住，摄取管道失败会在应用摄取管道之前捕获文档，当故障文档嵌套在新的故障中时，这会使修复变得更加复杂。模拟这些更改的最简单方法是使用 pipeline simulate API 或 simulate ingest API。
在摄取处理期间发生的故障将按照运行任何管道之前的状态进行存储。为了将文档重放到数据流中，我们需要为该文档重新运行所有适用的管道。
-
分离出要重放的故障
首先构建一个查询，该查询可用于一致地识别将要修复的故障。
POST my-datastream-ingest-example::failures/_search{ "query": { "bool": { "must": [ { "exists": { "field": "error.pipeline" } }, { "match": { "document.index": "my-datastream-ingest-example" } }, { "match": { "error.type": "illegal_argument_exception" } }, { "range": { "@timestamp": { "gt": "2025-05-01T00:00:00Z", "lte": "2025-05-02T00:00:00Z" } } } ] } } }- 要求
error.pipeline字段存在。这仅筛选摄取管道故障。 - 按数据流名称过滤，以修复指向特定索引的文档。
- 进一步缩小你试图修复的故障类型。在此示例中，我们针对特定类型的错误。
- 按时间戳过滤，仅检索某个时间点之前的故障。这提供了一组稳定的文档。
记录返回的文档。我们可以使用这些文档来模拟我们的修复逻辑是否合理
{ "took": 14, "timed_out": false, "_shards": { "total": 2, "successful": 2, "skipped": 0, "failed": 0 }, "hits": { "total": { "value": 1, "relation": "eq" }, "max_score": 2.575364, "hits": [ { "_index": ".fs-my-datastream-ingest-example-2025.05.16-000001", "_id": "cOnR2ZYByIwDXH-g6GpR", "_score": 2.575364, "_source": { "@timestamp": "2025-05-01T15:58:53.522Z", "document": { "index": "my-datastream-ingest-example", "source": { "@timestamp": "2025-05-01T00:00:00Z", "data": { "counter": 42 } } }, "error": { "type": "illegal_argument_exception", "message": "field [id] not present as part of path [data.id]", "stack_trace": """j.l.IllegalArgumentException: field [id] not present as part of path [data.id] at o.e.i.IngestDocument.getFieldValue(IngestDocument.java:202) at o.e.i.c.SetProcessor.execute(SetProcessor.java:86) ... 14 more """, "pipeline_trace": [ "my-datastream-default-pipeline" ], "pipeline": "my-datastream-default-pipeline", "processor_type": "set" } } } ] } }- 此文档将用于我们的模拟。
- 它有一个计数器值。
- 该文档缺少必填字段。
- 文档在
my-data-stream-default-pipeline中失败
- 要求
-
修复原始问题
由于摄取管道故障需要由其原始管道重新处理，因此在修复故障之前应解决这些管道中的任何问题。调查上面示例中提到的管道表明，有一个处理器期望某个字段存在，但该字段并不总是存在。
{ "my-datastream-default-pipeline": { "processors": [ { "set": { "field": "identifier", "copy_from": "data.id" } } ] } }- 预期
data.id字段存在。如果不存在，此管道将失败。
修复故障的根本原因通常是一个定制的过程。在此示例中，我们不会丢弃数据，而是将此标识符字段设为可选。
PUT _ingest/pipeline/my-datastream-default-pipeline{ "processors": [ { "set": { "field": "identifier", "copy_from": "data.id", "if": "ctx.data?.id != null" } } ] }- 仅当该字段存在时才条件性地运行处理器。
- 预期
-
创建一个管道以转换故障文档
我们必须将故障文档转换回其原始形式，并将其发送以重新处理。我们将创建一个管道来执行此操作
PUT _ingest/pipeline/my-datastream-remediation-pipeline{ "processors": [ { "script": { "lang": "painless", "source": """ ctx._index = ctx.document.index; ctx._routing = ctx.document.routing; def s = ctx.document.source; ctx.remove("error"); ctx.remove("document"); for (e in s.entrySet()) { ctx[e.key] = e.value; }""" } }, { "reroute": { "destination": "my-datastream-ingest-example" } } ] }- 将故障文档中的原始索引名称复制到文档的元数据中。如果你使用自定义文档路由，也请将其复制过来。
- 捕获原始文档的源。
- 丢弃
error字段，因为修复过程不需要它。 - 同时丢弃
document字段。 - 我们将原始文档源中的所有字段提取回文档的根目录。
- 由于失败的管道是
my-datastream-ingest-example上的默认管道，我们将使用reroute处理器将任何修复后的文档再次发送到该数据流的默认管道以进行重新处理。
-
测试你的管道
在将数据发送去重新索引之前，请务必使用示例文档测试相关管道以确保它们正常工作。首先，测试以确保修复管道生成的文档结构符合你的预期。我们可以为此使用 simulate pipeline API。
POST _ingest/pipeline/_simulate{ "pipeline": { "processors": [ { "script": { "lang": "painless", "source": """ ctx._index = ctx.document.index; ctx._routing = ctx.document.routing; def s = ctx.document.source; ctx.remove("error"); ctx.remove("document"); for (e in s.entrySet()) { ctx[e.key] = e.value; }""" } }, { "reroute": { "destination": "my-datastream-ingest-example" } } ] }, "docs": [ { "_index": ".fs-my-datastream-ingest-example-2025.05.16-000001", "_id": "cOnR2ZYByIwDXH-g6GpR", "_source": { "@timestamp": "2025-05-01T15:58:53.522Z", "document": { "index": "my-datastream-ingest-example", "source": { "@timestamp": "2025-05-01T00:00:00Z", "data": { "counter": 42 } } }, "error": { "type": "illegal_argument_exception", "message": "field [id] not present as part of path [data.id]", "stack_trace": """j.l.IllegalArgumentException: field [id] not present as part of path [data.id] at o.e.i.IngestDocument.getFieldValue(IngestDocument.java:202) at o.e.i.c.SetProcessor.execute(SetProcessor.java:86) ... 14 more """, "pipeline_trace": [ "my-datastream-default-pipeline" ], "pipeline": "my-datastream-default-pipeline", "processor_type": "set" } } } ] }- 上一步中编写的修复管道的内容。
- 我们在前面步骤中识别的示例故障文档的内容。
{ "docs": [ { "doc": { "_index": "my-datastream-ingest-example", "_version": "-3", "_id": "cOnR2ZYByIwDXH-g6GpR", "_source": { "data": { "counter": 42 }, "@timestamp": "2025-05-01T00:00:00Z" }, "_ingest": { "timestamp": "2025-05-01T20:58:03.566210529Z" } } } ] }- 索引已通过 reroute 处理器更新。
- 文档 ID 保持不变。
- 源应该完全匹配原始文档的内容。
既然修复管道已经测试完毕，请务必测试端到端摄取，以验证不会出现其他问题。为此，我们将使用 simulate ingestion API 来测试多个管道的执行。
POST _ingest/_simulate?pipeline=my-datastream-remediation-pipeline{ "pipeline_substitutions": { "my-datastream-remediation-pipeline": { "processors": [ { "script": { "lang": "painless", "source": """ ctx._index = ctx.document.index; ctx._routing = ctx.document.routing; def s = ctx.document.source; ctx.remove("error"); ctx.remove("document"); for (e in s.entrySet()) { ctx[e.key] = e.value; }""" } }, { "reroute": { "destination": "my-datastream-ingest-example" } } ] } }, "docs": [ { "_index": ".fs-my-datastream-ingest-example-2025.05.16-000001", "_id": "cOnR2ZYByIwDXH-g6GpR", "_source": { "@timestamp": "2025-05-01T15:58:53.522Z", "document": { "index": "my-datastream-ingest-example", "source": { "@timestamp": "2025-05-01T00:00:00Z", "data": { "counter": 42 } } }, "error": { "type": "illegal_argument_exception", "message": "field [id] not present as part of path [data.id]", "stack_trace": """j.l.IllegalArgumentException: field [id] not present as part of path [data.id] at o.e.i.IngestDocument.getFieldValue(IngestDocument.java:202) at o.e.i.c.SetProcessor.execute(SetProcessor.java:86) ... 14 more """, "pipeline_trace": [ "my-datastream-default-pipeline" ], "pipeline": "my-datastream-default-pipeline", "processor_type": "set" } } } ] }- 将管道设置为修复管道名称，否则将使用文档索引的默认管道。
- 前面步骤中修复管道的内容。
- 前面识别的示例故障文档的内容。
{ "docs": [ { "doc": { "_id": "cOnR2ZYByIwDXH-g6GpR", "_index": "my-datastream-ingest-example", "_version": -3, "_source": { "@timestamp": "2025-05-01T00:00:00Z", "data": { "counter": 42 } }, "executed_pipelines": [ "my-datastream-remediation-pipeline", "my-datastream-default-pipeline" ] } } ] }- 索引名称已更新。
- 默认管道运行后，源符合预期。
- 确保新的修复管道和原始的默认管道都已成功运行。
-
重新索引故障文档
将修复管道与故障存储区查询结合在 reindex 操作中以重放故障。
POST _reindex{ "source": { "index": "my-datastream-ingest-example::failures", "query": { "bool": { "must": [ { "exists": { "field": "error.pipeline" } }, { "match": { "document.index": "my-datastream-ingest-example" } }, { "match": { "error.type": "illegal_argument_exception" } }, { "range": { "@timestamp": { "gt": "2025-05-01T00:00:00Z", "lte": "2025-05-17T00:00:00Z" } } } ] } } }, "dest": { "index": "my-datastream-ingest-example", "op_type": "create", "pipeline": "my-datastream-remediation-pipeline" } }- 从故障存储区读取。
- 仅重新索引与我们要重放的文档相匹配的故障文档。
- 将目标设置为最初发送故障的数据流。
- 用修复管道替换管道。
{ "took": 469, "timed_out": false, "total": 1, "updated": 0, "created": 1, "deleted": 0, "batches": 1, "version_conflicts": 0, "noops": 0, "retries": { "bulk": 0, "search": 0 }, "throttled_millis": 0, "requests_per_second": -1, "throttled_until_millis": 0, "failures": [] }- 故障已修复。
提示既然在此数据流上启用了故障存储区，那么检查重新索引过程中是否存在任何进一步的故障是明智的。此过程中发生的故障最终可能会作为嵌套故障出现在故障存储区中。修复嵌套故障很快就会变成一件麻烦事，因为原始文档会在故障文档中嵌套多层深。因此，建议在没有其他故障发生的安静时期修复数据。此外，在执行修复之前滚动更新故障存储区可以更容易地丢弃任何新的嵌套故障，并且只对原始故障文档进行操作。
-
完成
一旦修复了任何故障，你可能希望从故障存储区中清除这些故障以释放空间，并避免关于已重放的失败数据的警告。否则，如果需要引用它们，你的故障将一直保留到最大故障存储区保留期限为止。
如前一节故障文档源所述，由于映射或索引问题而发生的故障将按照任何管道执行后的状态进行存储。这意味着要将文档重放到数据流中，我们需要确保跳过任何已经运行过的管道。
提示
通过将任何摄取管道编写为幂等的，你可以极大地简化此修复过程。在这种情况下，任何已经处理过并再次通过管道的文档都将保持不变。
-
分离出要重放的故障
首先构建一个查询，该查询可用于一致地识别将要修复的故障。
POST my-datastream-indexing-example::failures/_search{ "query": { "bool": { "must_not": [ { "exists": { "field": "error.pipeline" } } ], "must": [ { "match": { "document.index": "my-datastream-indexing-example" } }, { "match": { "error.type": "document_parsing_exception" } }, { "range": { "@timestamp": { "gt": "2025-05-01T00:00:00Z", "lte": "2025-05-02T00:00:00Z" } } } ] } } }- 要求
error.pipeline字段不存在。这会过滤掉所有摄取管道故障，并仅返回索引故障。 - 按数据流名称过滤，以修复指向特定索引的文档。
- 进一步缩小你试图修复的故障类型。在此示例中，我们针对特定类型的错误。
- 按时间戳过滤，仅检索某个时间点之前的故障。这提供了一组稳定的文档。
记录返回的文档。我们可以使用这些文档来模拟我们的修复逻辑是否合理。
{ "took": 1, "timed_out": false, "_shards": { "total": 1, "successful": 1, "skipped": 0, "failed": 0 }, "hits": { "total": { "value": 1, "relation": "eq" }, "max_score": 1.5753641, "hits": [ { "_index": ".fs-my-datastream-indexing-example-2025.05.16-000002", "_id": "_lA-GJcBHLe506UUGL0I", "_score": 1.5753641, "_source": { "@timestamp": "2025-05-02T18:53:31.153Z", "document": { "id": "_VA-GJcBHLe506UUFL2i", "index": "my-datastream-indexing-example", "source": { "processed": true, "data": { "counter": 37 } } }, "error": { "type": "document_parsing_exception", "message": "[1:40] failed to parse: data stream timestamp field [@timestamp] is missing", "stack_trace": """o.e.i.m.DocumentParsingException: [1:40] failed to parse: data stream timestamp field [@timestamp] is missing at o.e.i.m.DocumentParser.wrapInDocumentParsingException(DocumentParser.java:265) at o.e.i.m.DocumentParser.internalParseDocument(DocumentParser.java:162) ... 19 more Caused by: j.l.IllegalArgumentException: data stream timestamp field [@timestamp] is missing at o.e.i.m.DataStreamTimestampFieldMapper.extractTimestampValue(DataStreamTimestampFieldMapper.java:210) at o.e.i.m.DataStreamTimestampFieldMapper.postParse(DataStreamTimestampFieldMapper.java:223) ... 20 more """ } } } ] } }- 此文档将用于我们的模拟。
- 文档缺少必需的
@timestamp字段。 - 由于缺少时间戳，文档因
document_parsing_exception而失败。
- 要求
-
修复原始问题
可能存在广泛的索引故障。这些问题大多源于特定映射的值不正确。有时，大量新字段被动态映射，并且达到了最大映射字段数，因此无法再添加更多字段。在上面的示例中，正在索引的文档缺少必需的时间戳。
这些问题可能发生在多个地方：客户端发送的数据可能不完整，摄取管道可能没有产生正确的结果，或者可能需要更新索引映射以适应数据的变化。
一旦所有客户端和管道都生成了完整且正确的文档，并且你的映射已针对传入数据正确配置，请继续进行修复。
-
创建一个管道以转换故障文档
我们必须将故障文档转换回其原始形式，并将其发送以重新处理。我们将创建一个管道来执行此操作。由于示例故障是由于文档上没有时间戳，因此由于缺少原始时间戳，我们将直接使用故障发生时的时间戳。此方案假设我们正在修复的文档是在故障发生前后非常短的时间内创建的。如果不适用于你，你的修复过程可能需要进行调整。
PUT _ingest/pipeline/my-datastream-remediation-pipeline{ "processors": [ { "script": { "lang": "painless", "source": """ ctx._index = ctx.document.index; ctx._routing = ctx.document.routing; def s = ctx.document.source; ctx.remove("error"); ctx.remove("document"); for (e in s.entrySet()) { ctx[e.key] = e.value; }""" } } ] }- 将故障文档中的原始索引名称复制到文档的元数据中。如果你使用自定义文档路由，也请将其复制过来。
- 捕获原始文档的源。
- 丢弃
error字段，因为修复过程不需要它。 - 同时丢弃
document字段。 - 我们将原始文档源中的所有字段提取回文档的根目录。
@timestamp字段不会被覆盖，并且会存在于最终文档中。
重要提示请记住，在索引期间失败的文档已经由摄取处理器处理过了！除非你对管道进行了更改以修复原始问题，否则它不需要再次处理。确保应用于摄取管道的任何修复都在这里的管道逻辑中体现出来。
-
测试你的管道
在将数据发送去重新索引之前，请务必使用示例文档测试修复管道以确保其正常工作。最重要的是，确保修复管道生成的文档结构符合你的预期。我们可以为此使用 simulate pipeline API。
POST _ingest/pipeline/_simulate{ "pipeline": { "processors": [ { "script": { "lang": "painless", "source": """ ctx._index = ctx.document.index; ctx._routing = ctx.document.routing; def s = ctx.document.source; ctx.remove("error"); ctx.remove("document"); for (e in s.entrySet()) { ctx[e.key] = e.value; }""" } } ] }, "docs": [ { "_index": ".fs-my-datastream-indexing-example-2025.05.16-000002", "_id": "_lA-GJcBHLe506UUGL0I", "_score": 1.5753641, "_source": { "@timestamp": "2025-05-02T18:53:31.153Z", "document": { "id": "_VA-GJcBHLe506UUFL2i", "index": "my-datastream-indexing-example", "source": { "processed": true, "data": { "counter": 37 } } }, "error": { "type": "document_parsing_exception", "message": "[1:40] failed to parse: data stream timestamp field [@timestamp] is missing", "stack_trace": """o.e.i.m.DocumentParsingException: [1:40] failed to parse: data stream timestamp field [@timestamp] is missing at o.e.i.m.DocumentParser.wrapInDocumentParsingException(DocumentParser.java:265) at o.e.i.m.DocumentParser.internalParseDocument(DocumentParser.java:162) ... 19 more Caused by: j.l.IllegalArgumentException: data stream timestamp field [@timestamp] is missing at o.e.i.m.DataStreamTimestampFieldMapper.extractTimestampValue(DataStreamTimestampFieldMapper.java:210) at o.e.i.m.DataStreamTimestampFieldMapper.postParse(DataStreamTimestampFieldMapper.java:223) ... 20 more """ } } } ] }- 上一步中编写的修复管道的内容。
- 我们在前面步骤中识别的示例故障文档的内容。
{ "docs": [ { "doc": { "_index": "my-datastream-indexing-example", "_version": "-3", "_id": "_lA-GJcBHLe506UUGL0I", "_source": { "processed": true, "@timestamp": "2025-05-28T18:53:31.153Z", "data": { "counter": 37 } }, "_ingest": { "timestamp": "2025-05-28T19:14:50.457560845Z" } } } ] }- 索引已通过脚本处理器更新。
- 源应该反映所有修复，并匹配最终索引的预期文档结构。
- 在此示例中，我们发现故障时间戳保留在了源中。
-
重新索引故障文档
将修复管道与故障存储区查询结合在 reindex 操作中以重放故障。
POST _reindex{ "source": { "index": "my-datastream-indexing-example::failures", "query": { "bool": { "must_not": [ { "exists": { "field": "error.pipeline" } } ], "must": [ { "match": { "document.index": "my-datastream-indexing-example" } }, { "match": { "error.type": "document_parsing_exception" } }, { "range": { "@timestamp": { "gt": "2025-05-01T00:00:00Z", "lte": "2025-05-28T19:00:00Z" } } } ] } } }, "dest": { "index": "my-datastream-indexing-example", "op_type": "create", "pipeline": "my-datastream-remediation-pipeline" } }- 从故障存储区读取。
- 仅重新索引与我们要重放的文档相匹配的故障文档。
- 将目标设置为最初发送故障的数据流。示例中的修复管道将索引更新为正确的索引，但仍然需要目标。
- 用修复管道替换原始管道。这将防止任何默认管道运行。
{ "took": 469, "timed_out": false, "total": 1, "updated": 0, "created": 1, "deleted": 0, "batches": 1, "version_conflicts": 0, "noops": 0, "retries": { "bulk": 0, "search": 0 }, "throttled_millis": 0, "requests_per_second": -1, "throttled_until_millis": 0, "failures": [] }- 故障已修复。
提示既然在此数据流上启用了故障存储区，那么检查重新索引过程中是否存在任何进一步的故障是明智的。此过程中发生的故障最终可能会作为嵌套故障出现在故障存储区中。修复嵌套故障很快就会变成一件麻烦事，因为原始文档会在故障文档中嵌套多层深。因此，建议在不会出现其他故障的安静时期修复数据。此外，在执行修复之前滚动更新故障存储区可以更容易地丢弃任何新的嵌套故障，并且只对原始故障文档进行操作。
-
完成
一旦修复了任何故障，你可能希望从故障存储区中清除这些故障以释放空间，并避免关于已重放的失败数据的警告。否则，如果需要引用它们，你的故障将一直保留到最大故障存储区保留期限为止。