流程：每个服务器上都安装一个filebeat去读取日志文件存到kafka中，在日志服务器上安装logstash，logstash去kafka中读取数据存到elasticsearch中，最后在kibana上查看。
1，filebeat安装：
下载：https://www.elastic.co/cn/downloads/past-releases#filebeat
下载与elasticsearch版本一致，比如：filebeat-6.8.13-x86_64.rpm
yum install filebeat-6.8.13-x86_64.rpm -y
2，filebeat启动与关闭
systemctl start filebeat
systemctl stop filebeat
systemctl status filebeat //查看状态
ps -ef |grep filebeat
3，修改配置文件，将读取的内容存到kafka中，vim /etc/filebeat/filebeat.yml
修改：enabled: true #改为true
paths:
- /opt/test.log #要读取的文件路径
新增：
output.kafka:
hosts: ["10.10.21.175:9092"] #kafka地址
topic: "filebeat-systemlog-7-103" # 与filebeat的配置文件对应Kafka的topic名称
注释：
#-------------------------- Elasticsearch output ------------------------------
#output.elasticsearch: #注释掉此行
# Array of hosts to connect to.
#hosts: ["192.168.7.100:9200"] # 注释掉此行
4，改完记得重启一下
1，logstash安装：
wgethttps://artifacts.elastic.co/downloads/logstash/logstash-6.8.13.rpm
yum-y localinstall logstash-6.8.13.rpm
注意与elasticsearch版本一致
2，logstash启动与关闭
systemctl restart logstash
3，logstash从kafka上面采集日志数据
/etc/logstash/conf.d/目录下创建一个收集kafka配置文件,比如叫kafka-to-es.conf
内容为：
input {
kafka {
topics => "filebeat-systemlog-7-103" # 与filebeat的配置文件对应topic名称
bootstrap_servers => "10.10.21.175:9092" # kafka的地址
codec => "json"
}
}
output {
elasticsearch {
hosts => ["localhost:9200"] #es的地址
index => "kafka-%{+YYYY.MM.dd}" #es的索引名称
#user => "elastic"
#password => "changeme"
}
}
4，改完记得重启
