EFK实验架构图:
实现高并发,无需指定logstash
3台es+file,3台kafka
20.0.0.10 es+file
20.0.0.20 es+file
20.0.0.30 es+file
20.0.0.11 kafka
20.0.0.12 kafka
20.0.0.13 kafka
在es1主机上解压filebeat
cd filebeat
安装nginx服务
vim /usr/local/nginx/html/index.html
this is nginx
到浏览器测试一下页面访问
cp filebeat.yml filebeat.yml.bak
#做一个备份
vim filebeat.yml
- type: log
enabled: true
paths:
- /usr/local/nginx/logs/access.log
- /usr/local/nginx/logs/error.log
tags: ["nginx"]
fields:
service_name: 20.0.0.10_nginx
log_type: nginx
from: 20.0.0.10
output.kafka:
enabled: true
hosts: ["20.0.0.11:9092","20.0.0.12:9092","20.0.0.13:9092"]
topic: "nginx"
nohup ./filebeat -e -c filebeat.yml > filebeat.out &
#运行filebeat.yml
tail -f filebeat.out
#查看日志
在es1主机上配置logstash
cd /opt/log
vim kafka.conf
input {
kafka {
bootstrap_servers => "20.0.0.11:9092,20.0.0.12:9092,20.0.0.13:9092"
topics => "nginx"
type => "nginx_kafka"
codec => "json"
#解析json格式的代码
auto_offset_reset => "earliest"
#从头拉取,latest
decorate_events => true
#传递给es实例中的信息包含kafka的属性数据
}
}
output {
if "nginx" in [tags] {
elasticsearch {
hosts => ["20.0.0.20:9200","20.0.0.30:9200"]
index => "%{[fields][service_name]}-%{+YYYY.MM.dd}"
}
}
}
配置完成
使用logstash拉起kafka.conf
logstash -f kafka.conf --path.data /opt/nginx1
启动后到web页面查看是否生成节点文件
生成节点文件后到kibana页面创建索引
访问nginx的页面后到kibana页面查看数据是否同步
时间同步成功,实验完成。
elk和kafka如何配置可以去我的首页搜索即可