1.前言

ELK 一套非常出名的日志收集系统，不用多介绍了由Elasticsearch，Logstash，Kibana 组成。

然后采集数据的方式有很多，Log目录采集，tcp请求采集，redis，kafka，mq采集等。生产中多用kafka方式。所以这里也只介绍kafka采集方式。

1)Elasticsearch和Kibana搭建在前文已经介绍过了，地址：https://www.jianshu.com/p/0c0e124c1aa2

2)kafka搭建也介绍过了，地址：https://www.jianshu.com/p/79e29f653836

现在需要做的是把kafka的数据采集到es，用Logstash采集，过滤，解释数据

2.Logstash搭建

Logstash版本要和es版本一致
下载地址:https://www.elastic.co/cn/downloads/logstash
其它版本:https://www.elastic.co/cn/downloads/past-releases#logstash

image.png

下载解压，在config文件夹里参考logstash-sample格式新建一个文件kafkalog.conf,内容为:

input {
     kafka {
      topics => "kafkalog" #在kafka这个topics提取数据
      bootstrap_servers => "127.0.0.1:9092"  # kafka的地址
      codec => "json"  # 在提取kafka主机的日志时，需要写成json格式
            }
}
output {
  elasticsearch {
    hosts => ["127.0.0.1:9200"] #es地址
    index => "kafkalog" #把日志采集到es的索引名称
    #user => "elastic"
    #password => "changeme"
  }
}

#在logstash根目录执行启动命令
 bin/logstash -f config/kafkalog.conf

image.png 执行完成，也可以在浏览器上查看logstash信息 image.png

logstash安装成功！

3.测试结果

经过上面的步骤，ELK已经搭建完毕，现在需要往kafka的topics名“kafkalog”写日志，看es的index名为"kafkalog"有没有相关日志。
下面用上面搭建kafka时的.NET5程序往kafka写消息

image.png

向kafka发了6条消息

在kibana查看es image.png
image.png
看到es已经创建了一个kafkalog索引，里面刚好是刚才发的6条消息。 创建一个索引匹配规则 image.png
然后找到下面这个面板 image.png
image.png

ELK+kafka的搭建到这里就完成了。