情景：
我这边对接kafka，通过^符号来进行字符串分割
其中， 有个时间字段，但是，时间是14位的数字字符串
类似：20220517162218
于是，我这边需要转换成date类型的，才能在页面上按时间查询
转换成：2022-05-17 14:51:23
代码：
mutate{
split => ["message","^"]
add_field => {
"my_time" => "%{[message][8]}"
}
}
date {
match => ["my_time","yyyyMMddHHmmss"]
target => "my_time"
}
ruby {
code => "event.set('my_time', (event.get('my_time').time.localtime).strftime('%Y-%m-%d %H:%M:%S'))"
}
代码我不做过多的说明了，logstash工具的官方文档就那样，不好理解，只能自己试试。
完整的conf配置内容如下：
input{
file{
path => "C:/Users/Administrator/Desktop/logstash_7.12.0_kafka_test/alldata/*"
type => "default"
add_field => {"flag"=>"fk_11l_topic"}
start_position => "beginning"
type => "default"
codec => plain { charset => "UTF-8" }
}
}
filter {
mutate{
split => ["message","^"]
add_field => {
"my_time" => "%{[message][8]}"
}
}
date {
match => ["my_time","yyyyMMddHHmmss"]
target => "my_time"
}
ruby {
code => "event.set('my_time', (event.get('my_time').time.localtime).strftime('%Y-%m-%d %H:%M:%S'))"
}
if [message][0] {
mutate {
add_field => {
"acc_num" => "%{[message][0]}"
}
}
} else {
mutate {
add_field => {
"acc_num" => ""
}
}
}
if [message][8] {
mutate {
add_field => {
"violation_time" => "%{my_time}"
}
}
} else {
mutate {
add_field => {
"violation_time" => ""
}
}
}
mutate{
add_field => {
"scene_name" => "WHITELIST_MESSAGE"
"violation_type_code" => "21"
"kafka_topic" => "fk_111_topic"
}
remove_field=>["message"]
remove_field=>["@version"]
remove_field=>["my_time"]
}
}
output {
elasticsearch {
hosts => "127.0.0.1:9200"
index => "1212white111list_message_%{+YYYY-MM-dd}"
retry_on_conflict => 5
document_id => "%{acc_num}"
codec => plain { charset => "UTF-8" }
}
stdout { codec => rubydebug }
}
