Elasticsearch+Kafka整合

input {
  kafka {
    type => "level-one"
    auto_offset_reset => "smallest"
    codec => plain {
      charset => "GB2312"
    }
    group_id => "es"
    topic_id => "gamelog" //消息topic
    zk_connect => "zk-01:2181,zk-02:2181,zk-03:2181"
  }
}

filter {
  mutate {
    split => { "message" => "   " }
      add_field => {
        "event_type" => "%{message[3]}"
        "current_map" => "%{message[4]}"
        "current_X" => "%{message[5]}"
        "current_y" => "%{message[6]}"
        "user" => "%{message[7]}"
        "item" => "%{message[8]}"
        "item_id" => "%{message[9]}"
        "current_time" => "%{message[12]}"
     }
     remove_field => [ "message" ]
  } 
}

output {
    elasticsearch {
      index => "level-one-%{+YYYY.MM.dd}"
      codec => plain {
        charset => "utf-16be"//中文乱码问题
      }
      //ES的地址和端口
      hosts => ["elk-01:9200", "elk-02:9200", "elk-03:9200"]
    } 
}

C:\Users\dummy\Downloads\logstash-2.4.1\bin>.\logstash -f ..\conf\kafka-es.conf
Settings: Default pipeline workers: 8
Pipeline main started