如何在Elasticsearch中通过Filebeat自动关联索引实现嵌套数据插入？

可以直接使用Elasticsearch的Ingest Pipeline配合lookup处理器完成数据富集，全程不需要额外组件，具体步骤如下：

1. 创建数据富集的Ingest Pipeline #

假设存储玩家信息的索引名为players，我们创建一个名为enrich_players的流水线，自动将事件中的player1、player2字段替换为对应玩家的完整信息：

PUT _ingest/pipeline/enrich_players
{
  "description": "从players索引富集游戏事件中的玩家数据",
  "processors": [
    {
      "lookup": {
        "index": "players",
        "field": "player1",
        "match": "name.keyword",
        "target_field": "player1",
        "fetch_fields": ["age", "gender"],
        "ignore_missing": true
      }
    },
    {
      "lookup": {
        "index": "players",
        "field": "player2",
        "match": "name.keyword",
        "target_field": "player2",
        "fetch_fields": ["age", "gender"],
        "ignore_missing": true
      }
    }
  ]
}

参数说明： #

• index：指定存储玩家信息的目标索引
• field：传入事件中用于匹配的字段（即事件里的player1/player2的值）
• match：玩家索引中用于匹配的字段（需为keyword类型保证精确匹配）
• target_field：将匹配到的玩家数据写入的目标字段（直接覆盖原字符串字段）
• fetch_fields：需要从玩家索引中获取的额外字段
• ignore_missing：匹配不到玩家数据时不报错，保留原字段值

2. 确保玩家索引的字段可匹配 #

如果players索引中name字段是text类型，需要添加keyword子字段保证精确匹配：

PUT players/_mapping
{
  "properties": {
    "name": {
      "type": "text",
      "fields": {
        "keyword": {
          "type": "keyword",
          "ignore_above": 256
        }
      }
    }
  }
}

3. 配置Filebeat指定Pipeline #

修改Filebeat的配置文件，在Elasticsearch输出部分指定使用刚才创建的流水线：

output.elasticsearch:
  hosts: ["你的ES地址:9200"]
  index: "game-events-%{+yyyy.MM.dd}"
  pipeline: "enrich_players"  # 绑定富集流水线

配置完成后重启Filebeat，后续传入的事件会自动经过Elasticsearch的Ingest Pipeline处理，最终生成你预期的嵌套结构文档。

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