如何将Logstash接收的合并单条日志拆分后独立写入Elasticsearch

完全可以实现，核心使用Logstash内置的split过滤器即可将单条合并日志拆分为多条独立事件，最终分别写入Elasticsearch生成独立文档。

完整配置示例 #

# 输入部分按需调整为你实际的日志采集源，比如file、kafka等
input {
  # 示例用stdin方便测试，生产替换为实际输入源
  stdin {}
}

filter {
  # 清理原始日志首尾的数组括号
  mutate {
    gsub => [
      "message", "^\[", "",
      "message", "\]$", ""
    ]
  }

  # 将长字符串按`, `分割为包含多个logitem元素的数组
  mutate {
    split => { "message" => ", " }
  }

  # 核心：将数组每个元素生成独立的事件
  split {
    field => "message"
  }

  # 可选：提取logitem内的参数为独立字段，不需要可删除
  grok {
    match => { "message" => "logitem\(aaa=%{INT:aaa:int}, bbb=%{INT:bbb:int}, ccc=%{INT:ccc:int}\)" }
  }

  # 可选：清理不需要的冗余字段
  mutate {
    remove_field => [ "@version", "host" ]
  }
}

# 输出部分正常配置ES即可，拆分后的事件会独立写入
output {
  elasticsearch {
    hosts => ["http://你的ES地址:9200"]
    index => "你的索引名"
  }
  # 测试时可以加上stdout看拆分效果
  stdout {
    codec => rubydebug
  }
}

关键逻辑说明 #

• 前置的字符串清理和分割步骤，是为了把你示例中数组格式的日志字符串，转换为Logstash可识别的数组类型字段
• split过滤器是拆分多事件的核心，处理后每一条logitem都会成为完全独立的事件，后续过滤、输出规则都会针对单条事件执行
• 如果你的原始日志本身就是结构化的数组类型（而非数组字符串），可以省略前两步的字符串处理，直接调用split过滤器拆分对应数组字段即可

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