写入到kafka
input {
stdin { }
}
output {
kafka {
bootstrap_servers => "10.0.0.200:9092"
topic_id => "test"
}
}
##从kafka中将数据读出，写入到 elasticsearch中
input {
kafka {
bootstrap_servers => ["10.0.0.200:9092"] ##kafka地址，可以是集群
client_id => "test"
auto_offset_reset => "latest" ##从最新的偏移量开始消费
topics => ["test"] ## 数组类型，可配置多个topic
decorate_events => true #此属性会将当前topic、offset、group、partition等信息也带到message中
consumer_threads => 5
}
}
output {
elasticsearch {
hosts => ["10.0.0.140:9200"]
index => "kafka-%{+YYYY.MM.dd}"
}
stdout{
codec => rubydebug ##输出到屏幕上
}
}