ELK 7.x下Logstash解析JSON子元素导入ES配置求助

以下是满足你需求的完整Logstash配置，针对数组拆分和连续数字字段生成两个核心需求逐一实现：

完整配置代码 #

input {
  file {
    path => "/path/to/your/fruits.json" # 替换为你的JSON文件实际路径
    codec => "json"
    start_position => "beginning"
    sincedb_path => "/dev/null" # Linux/macOS用该值，Windows系统替换为"NUL"
  }
}

filter {
  # 拆分fruits.fruit数组，每个元素转为独立事件
  split {
    field => "[fruits][fruit]"
  }

  # 将数组元素的字段提升到顶层，清理冗余字段
  mutate {
    add_field => {
      "id" => "%{[fruits][fruit][id]}"
      "name" => "%{[fruits][fruit][name]}"
      "start" => "%{[fruits][fruit][start]}"
      "end" => "%{[fruits][fruit][end]}"
    }
    remove_field => ["fruits", "@version", "@timestamp"]
  }

  # 计算start到end的连续数字，生成逗号分隔的diffs字段
  ruby {
    code => '
      start_val = event.get("start").to_i
      end_val = event.get("end").to_i
      # 处理start大于end的边界情况，确保数字范围升序
      if start_val > end_val
        start_val, end_val = end_val, start_val
      end
      diffs_array = (start_val..end_val).to_a
      event.set("diffs", diffs_array.join(","))
    '
  }
}

output {
  elasticsearch {
    hosts => ["http://your-es-host:9200"] # 替换为你的Elasticsearch服务地址
    index => "fruits_index" # 替换为目标索引名称
    document_id => "%{id}" # 使用元素自身id作为ES文档ID
  }

  # 可选：输出到控制台验证处理结果
  stdout {
    codec => "rubydebug"
  }
}

配置细节说明 #

1. 数组拆分与独立文档生成 #

• split 插件会将fruits.fruit数组的每个元素拆分为独立的Logstash事件，直接满足"每个元素作为独立文档导入ES"的需求。
• 后续mutate插件将数组元素内的字段（id、name、start、end）提取到事件顶层，同时清理原fruits数组和Logstash默认生成的冗余字段，让最终文档结构更简洁。

2. 连续数字字段生成 #

• 使用ruby插件实现自定义逻辑：
  • 先将start和end字段转为整数，兼容JSON中字段为字符串类型的情况。
  • 处理start数值大于end的边界场景，确保生成的数字范围为升序。
  • 通过Ruby范围对象生成连续数字数组，再用join(",")转换为逗号分隔的字符串，存入diffs字段。

3. ES文档ID指定 #

• 在elasticsearch输出插件中，通过document_id => "%{id}"直接引用事件中的id字段，将其作为Elasticsearch文档的唯一ID。

示例验证 #

假设你的输入JSON片段如下：

{
  "fruits": {
    "fruit": [
      {
        "id": "fruit_001",
        "name": "apple",
        "start": 2,
        "end": 5
      },
      {
        "id": "fruit_002",
        "name": "banana",
        "start": 7,
        "end": 3
      }
    ]
  }
}

导入Elasticsearch后的最终文档格式为：

{
  "_id": "fruit_001",
  "_index": "fruits_index",
  "_source": {
    "id": "fruit_001",
    "name": "apple",
    "start": "2",
    "end": "5",
    "diffs": "2,3,4,5"
  }
}

{
  "_id": "fruit_002",
  "_index": "fruits_index",
  "_source": {
    "id": "fruit_002",
    "name": "banana",
    "start": "7",
    "end": "3",
    "diffs": "3,4,5,6,7"
  }
}

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