如何修改Logstash配置将单行多记录解析为指定格式CSV？

问题背景 #

现有如下Logstash配置：

input {
file{
    path => "/Users/.../Work/Projects/ELK/Logstash/Input/*.txt"
    start_position => beginning
    codec => "json"
    type => "data"
    sincedb_path => "NUL"
   }
}

filter { 
    grok { match => { "message" => "<%{NUMBER}>%{SYSLOGTIMESTAMP} %{IPV4} %{HOSTNAME:nodeName}: %{NUMBER} %{WORD} \[%{DATA}\]: %{GREEDYDATA:[anotherField]} \n" } }
        # Create array of strings
        mutate { split => { "[anotherField]" => "| " } }
        # Create a separate event for each array entry
        split { field => "[anotherField]" } 
    }

output {
stdout{}
 file {
   path => "/Users/...../Work/Projects/ELK/Logstash/test.csv"
   codec => line { format => "%{nodeName},%{anotherField}"}
 }
}

输入是包含多条记录的单行JSON数据：

{"@timestamp":"2022-12-01T13:30:00.004Z","message":"<190>Dec  1 14:29:59 10.62.161.199 AA-AMG3U: 0950198238 NN [MDA 8/4]: LN44 SA 2022 Dec  1 14:29:59:87 CET 17 4001 10.XX.133.XX 56560 401 91.235.10.25 15179 2400160261XXXXX_467000XXXX1_35292011220XXXXX_string1 | LN44 SD 2022 Dec  1 14:29:59:89 CET 17 4001 10.XX.133.XX 56560 401 91.235.10.25 15179 2400160261XXXXX_467000XXXX1_35292011220XXXXX_string2 | LN44 SA 2022 Dec  1 14:29:59:87 CET 17 4001 10.XX.133.XX 56560 401 91.235.10.25 15179 2400160261XXXXX_467679XXXX2_35292011220XXXXX_string1 \n","@version":"1","host":"100.62.161.XXX"}

当前输出是将每条记录作为整体的CSV：

AA-AMG3U,LN44 SA 2022 Dec  1 14:29:59:87 CET 17 4001 10.XX.133.XX 56560 401 91.235.10.25 15179 2400160261XXXXX_467000XXXX1_35292011220XXXXX_string1 
AA-AMG3U,LN44 SD 2022 Dec  1 14:29:59:89 CET 17 4001 10.XX.133.XX 56560 401 91.235.10.25 15179 2400160261XXXXX_467000XXXX1_35292011220XXXXX_string2 
AA-AMG3U,LN44 SA 2022 Dec  1 14:29:59:87 CET 17 4001 10.XX.133.XX 56560 401 91.235.10.25 15179 2400160261XXXXX_467679XXXX2_35292011220XXXXX_string1

需要将输出修改为指定的字段拆分后的CSV格式：

AA-AMG3U,LN44,SA,2022 Dec  1 14:29:59:87 CET,17,4001,10.XX.133.XX 56560,401 91.235.10.25 15179,2400160261XXXXX,467000XXXX1,35292011220XXXXX,string1 
AA-AMG3U,LN44,SD,2022 Dec  1 14:29:59:87 CET,17,4001,10.XX.133.XX 56560,401 91.235.10.25 15179,2400160261XXXXX,467000XXXX1,35292011220XXXXX,string1 
AA-AMG3U,LN44,SA,2022 Dec  1 14:29:59:87 CET,17,4001,10.XX.133.XX 56560,401 91.235.10.25 15179,2400160261XXXXX,467000XXXX2,35292011220XXXXX,string1

修改后的Logstash配置 #

input {
  file{
      path => "/Users/.../Work/Projects/ELK/Logstash/Input/*.txt"
      start_position => beginning
      codec => "json"
      type => "data"
      sincedb_path => "NUL"
     }
}

filter { 
    # 解析syslog结构，提取nodeName和原始多记录内容
    grok { 
        match => { "message" => "<%{NUMBER}>%{SYSLOGTIMESTAMP} %{IPV4} %{HOSTNAME:nodeName}: %{NUMBER} %{WORD} \[%{DATA}\]: %{GREEDYDATA:rawRecords} \n" } 
    }
    
    # 拆分多记录为数组，并清理每条记录前后的空白
    mutate { 
        split => { "rawRecords" => "| " } 
        strip => ["rawRecords"]
    }
    
    # 为数组中每条记录生成独立事件
    split { 
        field => "rawRecords" 
    }
    
    # 解析单条记录的基础字段
    grok {
        match => {
            "rawRecords" => "%{WORD:prefix} %{WORD:type} %{DATA:timestamp} %{NUMBER:num1} %{NUMBER:num2} %{IPV4:srcIp} %{NUMBER:srcPort} %{NUMBER:status} %{IPV4:dstIp} %{NUMBER:dstPort} %{DATA:combinedField}"
        }
    }
    
    # 拆分下划线连接的复合字段为独立字段
    mutate {
        split => { "combinedField" => "_" }
        add_field => {
            "field1" => "%{[combinedField][0]}"
            "field2" => "%{[combinedField][1]}"
            "field3" => "%{[combinedField][2]}"
            "field4" => "%{[combinedField][3]}"
        }
        # 清理临时字段与冗余原始字段
        remove_field => ["rawRecords", "combinedField", "message", "@version", "host"]
    }
}

output {
    stdout{}
    file {
       path => "/Users/...../Work/Projects/ELK/Logstash/test.csv"
       codec => line { 
           format => "%{nodeName},%{prefix},%{type},%{timestamp},%{num1},%{num2},%{srcIp} %{srcPort},%{status} %{dstIp} %{dstPort},%{field1},%{field2},%{field3},%{field4}"
       }
    }
}

配置说明 #

1. 字段重命名与清理：将原始复合记录字段命名为rawRecords，并用strip去除拆分后记录的前后空白，避免解析错误；
2. 单记录解析：新增grok规则，把每条独立记录拆分为前缀、类型、时间戳等基础字段；
3. 复合字段拆分：将下划线连接的长字段拆分为4个独立字段，匹配目标CSV的格式要求；
4. 输出格式化：在output的line codec中按目标顺序拼接所有字段；
5. 冗余清理：移除临时字段和不需要的原始字段，精简输出内容。

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