将kafka数据导入Elasticsearch

基于当前最新版本logstash 6.0.0 不同版本的logstash输入输出语法略有不同。

Kafka->logstash->elasticsearch->kibana

下载logstash并解压

进入logstash/config目录新建文件 kafka-logstash-es.conf

编写文件kafka-logstash-es.conf

input {    //日志写入的数据源
kafka {
bootstrap_servers => ["ip:9092"]  //kafka broker
group_id => "1"
topics => ["test-kafka-elasticsearch"]    //kafka topic
consumer_threads => 5
decorate_events => true
codec => "json"        //数据格式
}
}

output { //日志数据写出
elasticsearch {
hosts => ["ip:9200"]  //Elasticsearch所在
index => "test"                 //输出索引
codec => "json"                //输出格式
}
}

启动命令 bin/logstash -f /logstash安装位置/logstash-6.0.0/config/kafka-logstash-es.conf &

启动成功后向kafka写入数据

登录kibana查看数据

kibana使用可查看kibana使用