下面介紹兩個Logstash的配置Demo：

Demo1：

input {
kafka {
  zk_connect => "10.10.16.2:2181,10.10.16.3:2181,10.10.16.4:2181"
  group_id => "test-consumer-group"
  topic_id => "MyPattern"
  reset_beginning => false # boolean (optional)， default: false
  consumer_threads => 5  # number (optional)， default: 1
  decorate_events => true # boolean (optional)， default: false
  }
}

filter{
    mutate{

       split => ["message",","]
    }
    #第一個資料的內容中ORA-xxxxx這種格式，則這條內容是ora錯誤。新增二個欄位
    mutate{
        add_field =>   {
                "SRC_ADDRESS" => "%{[message][0]}"
                "DEST_ADDRESS" => "%{[message][1]}"
                "SRC_PORT" => "%{[message][2]}"
                "DEST_PORT" => "%{[message][3]}"
                "TRANS_PROTOCOL" => "%{[message][4]}"
                "PACKETS" => "%{[message][5]}"
                "BYTES" => "%{[message][6]}"
                "FLAGS" => "%{[message][7]}"
                "START_TIME" => "%{[message][8]}"
   }
}

}
 

Demo2：

input {
 file {
        path => ["/home/test.csv"]
        #type => "system"
       #start_position => "beginning"
    }
}

filter{
                mutate{
                        gsub => [ "message", "\r", "" ]
                }

                mutate {
                        split => ["message",","]
                }

                mutate{
                add_field =>   {
                                        "id" => "%{[message][0]}"
                                        "time" => "%{[message][1]}"
                                        "userId" => "%{[message][2]}"
                                        "pc" => "%{[message][3]}"
                                        "stat" => "%{[message][4]}"
                }
        }
}

output {
    stdout { codec => rubydebug }
}