logstash 日誌 kafka 到 ES

vim newgames.conf

input{
    kafka {
        bootstrap_servers => "10.4.22.38:9099,10.4.22.39:9099,10.4.22.67:9099"
        topics => ["newgameagby", "newgameagzr", "newgameavia", "newgameebetzr", "newgameimdj", "newgameimqp", "newgameimty", "newgameppdz", "newgamesgcp", "newgametcgcp", "newgamexjty","newgamelhdj","newgamepgdz","apidata-ybqp","newgameybqp","newgameybcp","newgamewmzr"]
        group_id => "newgame2"
        auto_offset_reset => "earliest"
        codec => "json"
        consumer_threads => 30
   }
}

filter {
  mutate { add_field => { "[@metadata][index_name]" => "%{es_index_name}" } }
  mutate { remove_field => ["es_index_name"] }
}

output {
    stdout {
        codec => rubydebug
    }
    elasticsearch {
        index  =>  "%{[@metadata][index_name]}"
        document_id => "%{id}"
        document_type  => "_doc"
        hosts => ["10.4.22.38:9200","10.4.22.39:9200","10.4.22.67:9200"]
        codec => json
    }
}