如何在Logstash中解析日志文件中的JSON对象数组并将每个数组元素拆分为独立文档发送至Elasticsearch

刚好我之前处理过类似的日志拆分需求，给你两种可行的解决方案，你可以根据自己的需求选择：

这种方案会完整保留日志的原始时间戳，同时将顶级JSON数组里的每个对象拆成独立的Elasticsearch文档，是生产环境更常用的方式。以下是完整的Logstash配置：

input {
  # 根据你的实际输入源调整，比如file、beats、tcp等
  file {
    path => "/path/to/your/target/logs/*.log"
    start_position => "beginning"
    sincedb_path => "/dev/null" # 调试时用，避免重复读取
  }
}

filter {
  # 1. 用grok提取日志开头的时间戳和后续的JSON内容
  grok {
    match => { "message" => "^%{TIMESTAMP_ISO8601:log_timestamp} %{GREEDYDATA:json_payload}$" }
    # 把提取到的日志时间设置为事件的@timestamp（ES会用这个字段做时间索引）
    add_field => { "@timestamp" => "%{log_timestamp}" }
    # 移除冗余的原始message字段
    remove_field => ["message"]
  }

  # 2. 解析提取到的JSON payload
  json {
    source => "json_payload"
    target => "parsed_json"
    # 解析完成后删除原始的json_payload字段
    remove_field => ["json_payload"]
  }

  # 3. 拆分顶级JSON数组为独立事件
  split {
    field => "parsed_json"
    # 将数组元素直接放到事件根级别，更符合ES文档的结构
    target => ""
  }

  # 可选：如果grok的TIMESTAMP_ISO8601对微秒的解析有问题，用date过滤器手动转换
  date {
    match => ["log_timestamp", "yyyy-MM-dd HH:mm:ss,SSSSSS"]
    target => "@timestamp"
  }
}

output {
  elasticsearch {
    hosts => ["http://your-es-cluster:9200"]
    index => "logstash-custom-%{+YYYY.MM.dd}" # 按日期拆分索引
  }
  # 调试用：在控制台输出处理后的事件结构，确认拆分效果
  stdout { codec => rubydebug }
}

关键步骤说明： #

• grok提取：精准匹配日志开头的时间戳和后面的JSON内容，把两部分分开处理；
• JSON解析：将提取到的JSON字符串转成Logstash可操作的结构化数据；
• split拆分：指定顶级数组字段，Logstash会自动把数组里的每个对象拆成独立事件；
• 时间戳校准：确保ES里存储的日志时间和原始日志一致，避免用Logstash的接收时间。

如果你确实不需要保留原始日志时间戳（不推荐），可以先移除开头的时间戳内容，然后利用官方提到的JSON codec自动拆分顶级数组的特性：

input {
  file {
    path => "/path/to/your/target/logs/*.log"
    start_position => "beginning"
  }
}

filter {
  # 用正则移除日志开头的时间戳部分（匹配数字、连字符、冒号、逗号、空格的组合）
  mutate {
    gsub => ["message", "^[0-9\-:, ]+", ""]
  }
}

output {
  elasticsearch {
    hosts => ["http://your-es-cluster:9200"]
    index => "logstash-simple-%{+YYYY.MM.dd}"
    # 启用JSON codec，自动拆分顶级JSON数组为独立事件
    codec => json
  }
  stdout { codec => rubydebug }
}

注意事项： #

• 调试时一定要打开stdout { codec => rubydebug }，查看每个阶段的事件结构，方便定位问题；
• 如果你的日志路径或输入源有特殊配置，记得调整input部分的参数；
• 若ES有认证需求，需要在elasticsearch输出里添加user和password参数。

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