使用Logstash实现Kafka到Elasticsearch的数据拆分索引配置

一、需求与数据格式说明 #

从Kafka接收的原始数据结构：

{
  "id": "XYZ",
  "index": "original_data_index",
  "updated_data": {
    "id": "XYZ1",
    "index": "updated_data_index"
  }
}

需实现：

• 顶层核心数据写入Elasticsearch的index1，结构如下：

{
  "id": "XYZ",
  "index": "original_data_index"
}

• updated_data字段下的内容写入Elasticsearch的index2，结构如下：

{
  "id": "XYZ1",
  "index": "updated_data_index"
}

二、Logstash管道配置实现 #

通过clone插件复制事件、mutate插件过滤字段，结合条件判断将事件路由到对应Elasticsearch索引，完整配置示例：

input {
  kafka {
    bootstrap_servers => "你的Kafka broker地址:9092"
    topics => ["目标主题名"]
    codec => json # 自动解析Kafka中的JSON数据
  }
}

filter {
  # 复制一份事件，用于处理updated_data内容
  clone {
    clones => ["updated_event"]
  }

  # 处理原始事件（写入index1）：移除updated_data及冗余元字段
  if [type] != "updated_event" {
    mutate {
      remove_field => ["updated_data", "@timestamp", "@version"]
    }
  }

  # 处理复制出的事件（写入index2）：替换为updated_data内容并重新解析
  if [type] == "updated_event" {
    mutate {
      replace => { "message" => "%{[updated_data]}" }
      remove_field => ["updated_data", "id", "index", "@timestamp", "@version"]
    }
    json {
      source => "message"
      remove_field => ["message"]
    }
  }
}

output {
  # 写入原始数据到index1
  if [type] != "updated_event" {
    elasticsearch {
      hosts => ["你的ES节点地址:9200"]
      index => "index1"
      # 可选：指定文档ID实现幂等写入
      # document_id => "%{id}"
    }
  }

  # 写入更新后数据到index2
  if [type] == "updated_event" {
    elasticsearch {
      hosts => ["你的ES节点地址:9200"]
      index => "index2"
      # 可选：指定文档ID实现幂等写入
      # document_id => "%{id}"
    }
  }

  # 可选：调试用输出到控制台
  stdout {
    codec => rubydebug
  }
}

三、最佳实践与注意事项 #

• 字段精简：移除业务无关字段及Logstash元字段（如@timestamp），降低ES存储压力，提升查询效率。
• 幂等写入：用数据中的id作为ES的document_id，避免重复消息导致的文档重复创建。
• 错误处理：在ES输出中开启dead_letter_queue.enable => true，将写入失败的事件存入死信队列，便于后续排查重试。
• 性能调优：
  • 调整pipeline.batch.size和pipeline.batch.delay参数，适配Kafka消息量与ES性能，提升批量写入吞吐量。
  • 配置bulk_max_bytes控制ES批量请求大小，避免过大请求压垮ES节点。
• 数据校验：在json插件中开启validate => true，确保Kafka传入的JSON格式合法，避免格式错误中断处理。
• 索引模板预定义：提前为index1和index2创建ES索引模板，定义字段类型、分词器等配置，避免自动映射导致的字段类型不匹配问题。

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