1，需求说明

===原始数据格式： 1条 （2*2）==》4个指标数据
[{"app":"aa","url":"www.1.com","metrics":[{"name":"cpu","value":11},{"name":"mem","value":1}]},
{"app":"bb","url":"www.2.com","metrics":[{"name":"cpu","value":12},{"name":"mem","value":2}]}]====> 希望得到的数据格式 =》 4个指标数据单独为一行
1{"name" => "cpu","app" => "aa","@timestamp" => 2024-11-22T05:57:29.671882657Z,"url" => "www.1.com","value" => 11
}
2{"name" => "mem","app" => "aa","@timestamp" => 2024-11-22T05:57:29.671882657Z,"url" => "www.1.com","value" => 1
}
3{"name" => "cpu","app" => "bb","@timestamp" => 2024-11-22T05:57:29.672029155Z,"url" => "www.2.com","value" => 12
}
4{"name" => "mem","app" => "bb","@timestamp" => 2024-11-22T05:57:29.672029155Z,"url" => "www.2.com","value" => 2
}

2，logstash 任务配置

[root@t1 test]# cat a.json
[{"app":"aa","url":"www.1.com","metrics":[{"name":"cpu","value":11},{"name":"mem","value":1}]},{"app":"bb","url":"www.2.com","metrics":[{"name":"cpu","value":12},{"name":"mem","value":2}]}][root@t1 test]# cat logstash-pipline.conf
input {#    kafka {#        bootstrap_servers => "localhost:9092"#        topics => ["your_topic_name"]#        codec => "json"#        add_field => {#            "source" => "kafka"#        }#    }file {path => "/root/test/a.json"  # 指定JSON文件的路径start_position => "beginning"  # 从文件的开头开始读取（可选，默认为"end"，即从文件末尾开始读取新添加的内容）codec => "json"  # 指定文件编码格式为JSONsincedb_path => "/dev/null"  # 禁用sincedb文件，以便每次运行时都重新读取整个文件（可选，通常用于调试）}
}filter {if [message] { ##拆分message: 大数组json {source => "message"target => "data_array"}split {field => "data_array"remove_field => "message"}}split { ##拆分metrics: 小数组field => "metrics"}mutate {add_field => {"field1" => "%{metrics}"}}json {source => "field1"}mutate {remove_field => ["field1","metrics","log","host","@version"] #log,host,@version是系统添加的字段}
}output {stdout { codec => rubydebug }  # 将数据输出到控制台，并使用rubydebug编解码器进行格式化
}[root@t1 test]# /data01/logstash-8.7.1/bin/logstash -f /root/test/logstash.conf