logstash filter-split 示例

input {
file {
path => "/opt/logstash/config/aa.log" #配置读取的文件
start_position => "beginning" #从文件开始位置读取
discover_interval => 5 #设置logstash读取新文件的时间间隔
max_open_files => 10 #配置当前input可以监控的文件的最大值
close_older => 3600 #结束时间，即如果在限制时间段内没有更新内容，就关闭监听它的文件句
柄
sincedb_path => "/data/sincedb_test.txt" #记录读取的位置
sincedb_write_interval => 15
codec => json { #配置文本类型
charset => "UTF-8"
}
}
}

filter {

if "M00002" in [message] {
mutate {
split => ["message", "|"] #原始日志按"|"切割

    add_field => {               #增加字段，对字段命名
      "timestamp" => "%{message[0]}"
      "thread" => "%{message[1]}"
      "loglevel" => "%{message[2]}"
      "class" => "%{message[3]}"
      "aa" => "%{message[4]}"
      "bb" => "%{message[5]}"
      "modelid" => "%{message[6]}"
      "cc" => "%{message[7]}"
      "dd" => "%{message[8]}"
      "ee" => "%{message[9]}"
      "ff" => "%{message[10]}"
   }

   remove_field => ["message"]   #删除原始字段
}
   }

   else {
        mutate {
    split => ["message", "|"]  #原始日志按"|"切割

    add_field => {               #增加字段，对字段命名
      "timestamp" => "%{message[0]}"
      "thread" => "%{message[1]}"
      "loglevel" => "%{message[2]}"
      "class" => "%{message[3]}"
      "aa" => "%{message[4]}"
      "bb" => "%{message[5]}"
      "modelid" => "%{message[6]}"
      "cc" => "%{message[7]}"
      "dd" => "%{message[8]}"
      "ee" => "%{message[9]}"
      "ff" => "%{message[10]}"
   }

   remove_field => ["message"]   #删除原始字段
}
 }
date { # 日期格式化
       match => ["timestamp", "ISO8601"]
   }

}

output {
file {
path => "/opt/logstash/config/bb.txt" #输出到一个文件内
}
stdout{codec => rubydebug}
}