input {

kafka {

bootstrap_servers => ["192.168.1.1:9092"] # 注意这里配置的kafka的broker地址不是zk的地址

group_id => "logstash" # 自定义groupid

topics => ["ttt.log.topic"] # kafka topic 名称

consumer_threads => 5

decorate_events => true

codec => "json"

}

}

output {

elasticsearch { hosts => ["localhost:9200"] }

stdout { codec => rubydebug }

}