Filebeat+Kafka+ELK搭建

news2024/11/6 3:13:44

在这里插入图片描述

---------------- Filebeat+Kafka+ELK ----------------
1.部署 Zookeeper+Kafka 集群（前面已经配过 20.0.0.101、20.0.0.102、20.0.0.103）

https://blog.csdn.net/m0_56509725/article/details/132908696?spm=1001.2014.3001.5501

1.1 配置ELK 在：

https://blog.csdn.net/m0_56509725/article/details/132853050?spm=1001.2014.3001.5501

2.部署 Filebeat

cd /usr/local/filebeat

vim filebeat.yml
filebeat.prospectors:
- type: log
  enabled: true
  paths:
    - /var/log/httpd/access_log
  tags: ["access"]
  
- type: log
  enabled: true
  paths:
    - /var/log/httpd/error_log
  tags: ["error"]
  
......
#添加输出到 Kafka 的配置
output.kafka:
  enabled: true
  hosts: ["20.0.0.101:9092","20.0.0.102:9092","20.0.0.103:9092"]    #指定 Kafka 集群配置
  topic: "httpd"    #指定 Kafka 的 topic
 #注意要把Elasticsearch output下的内容注释掉，否则会报错

在这里插入图片描述

#启动 filebeat

./filebeat -e -c filebeat.yml

在这里插入图片描述

3.部署 ELK，在 Logstash 组件所在节点上新建一个 Logstash 配置文件

cd /etc/logstash/conf.d/

vim kafka.conf
input {
    kafka {
        bootstrap_servers => "20.0.0.101:9092,20.0.0.102:9092,20.0.0.103:9092"  #kafka集群地址
        topics  => "httpd"     #拉取的kafka的指定topic
        type => "httpd_kafka"  #指定 type 字段
        codec => "json"        #解析json格式的日志数据
		auto_offset_reset => "latest"  #拉取最近数据，earliest为从头开始拉取
		decorate_events => true   #传递给elasticsearch的数据额外增加kafka的属性数据
    }
}

output {
  if "access" in [tags] {
    elasticsearch {
      hosts => ["20.0.0.105:9200"]
      index => "httpd_access-%{+YYYY.MM.dd}"
    }
  }
  
  if "error" in [tags] {
    elasticsearch {
      hosts => ["20.0.0.105:9200"]
      index => "httpd_error-%{+YYYY.MM.dd}"
    }
  }
  
  stdout { codec => rubydebug }
}