| ip | host | 角色 |
| 192.168.73.156 | server1 | ES |
| 192.168.73.157 | server2 | ES |
| 192.168.73.158 | server3 | ES |
| 192.168.73.159 | server4 | cerebro |
| 192.168.73.160 | server5 | logstash filebeat |
| 192.168.73.161 | server6 | kafka |
| 192.168.73.162 | server7 | kafka |
| 192.168.73.163 | server8 | kafka |
目录
ES节点部署filebeat
kafka
集群部署
三台主机安装 Kafka
server6节点生成储目录唯一的 UUID
三台节点启动 Kafka
Kafka 集群可用性验证
Kafka 集群任意节点创建 Topic
Kafka 集群任意节点查看 Topic
Kafka 集群任意节点生产者测试
Kafka 集群任意节点查看主题
Kafka 集群任意节点查看 topic 详细信息
FIlebeat 配置文件修改
配置kafka输出
Kafka 集群任意节点消费者测试
Logstash 消费 Topic 消息
ES节点部署filebeat
三台ES节点部署filebeat插件
[root@server1 ~]# rpm -ivh filebeat-7.6.1-x86_64.rpm
[root@server1 ~]# cd /etc/filebeat/
[root@server1 filebeat]# cd modules.d/
一定要把对应的提起来，不然用不了，都是默认disable
[root@server1 modules.d]# filebeat modules enable elasticsearch
Enabledelasticsearch
[root@server1 modules.d]# vim elasticsearch.yml
特别注意语法缩进问题，一定要缩进两格
参考网址：Elasticsearch module | Filebeat Reference [7.6] | Elastic https://www.elastic.co/guide/en/beats/filebeat/7.6/filebeat-module-elasticsearch.html
[root@server1 modules.d]# cd ..
[root@server1 filebeat]# vim filebeat.yml
三台ES统一输出到同一台节点收集日志
这里注意把密码和用户开启，因为设置了xpack认证，没有密码的话kibana识别不到日志
[root@server1 filebeat]# systemctl enable --now filebeat.service
将server1的配置文件同步拷贝给server2 server3
[root@server1 modules.d]# scp elasticsearch.yml server2:/etc/filebeat/modules.d/
[root@server1 modules.d]# scp elasticsearch.yml server3:/etc/filebeat/modules.d/
其余两台节点同步修改filebeat.yml 文件，配置输出节点，开启用户和密码
打开kibana网页，已经检索到了日志同步输出
查看指定主机日志
kafka
Kafka 作为一个分布式平台，以容错、高可用性的集群形式运行。其核心架构包含以下几个关键概念：
- 主题（Topic）：Kafka 对消息的高层抽象，类似于数据库中的表。每个主题维护一个分区的日志。
- 生产者（Producer）：负责向 Kafka 的特定主题发布消息。
- 消费者（Consumer）：向主题注册并接收消息。消费者通过记录偏移量（Offset）来追踪自己读取到了日志的哪个位置。
- 代理（Broker）：Kafka 集群中的服务器节点。主题被分为多个分区（Partition）分布在不同 Broker 上，以实现数据的水平扩展和负载均衡。
集群部署
三台节点一定要配置解析
vim /etc/hosts
- 语言依赖：Apache Kafka 的核心代码主要是用Scala和Java编写的。
- JVM 虚拟机：Java 程序不能直接在操作系统（Linux/Windows）上跑，它必须运行在 Java 虚拟机（JVM）上。JDK（Java Development Kit）包含了 JRE（运行环境）和 JVM。
rpm -ivh jdk-17.0.17_linux-x64_bin.rpm
java -version
三台主机安装 Kafka
[root@server6 ~]# tar zxf kafka_2.12-3.9.1.tgz -C /opt
[root@server6 ~]#cd /opt/
[root@server6 ~]#mv kafka_2.12-3.9.1/ kafka
[root@server6 ~]# cd /opt/kafka/config/kraft/
先备份一份原始文件，防止改错后无法恢复。
[root@server6 ~]#cp server.properties{,.bak}
[root@server6 ~]#mkdir /opt/kafka/logs[root@server6 ~]#vim server.properties
将配置文件同步拷贝给server7,server8
[root@server6 ~]#scp server.properties server7:/opt/kafka/config/kraft/
[root@server6 ~]#scp server.properties server8:/opt/kafka/config/kraft/
server7 server8 做相应修改
提示：另外两台机器（server7, server8）只需修改node.id和listeners中的 IP 地址，其余保持一致。
server7为162 server8 为163
server6节点生成储目录唯一的 UUID
Kafka KRaft 模式不需要 Zookeeper，但需要先生成一个唯一的集群 ID 并初始化目录。在任意一台机器上执行：
[root@server6 ~]# KAFKA_CLUSTER_ID="$(/opt/kafka/bin/kafka-storage.sh random-uuid)"
[root@server6 ~]# echo ${KAFKA_CLUSTER_ID}
mdfgkhX1Q6u2iBJpq15EYw
三个节点使用该UUID格式化
/opt/kafka/bin/kafka-storage.sh format -t mdfgkhX1Q6u2iBJpq15EYw -c /opt/kafka/config/kraft/server.properties
三台节点启动 Kafka
/opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/kraft/server.properties
Kafka 集群可用性验证
为了证明集群搭建成功，我们需要创建一个 Topic 并测试消息发送。
Kafka 集群任意节点创建 Topic
[root@server6 ~]#/opt/kafka/bin/kafka-topics.sh --create --topic testtopics --bootstrap-server localhost:9092 --replication-factor 3 --partitions 3
Kafka 集群任意节点查看 Topic
[root@server7 kraft]# /opt/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:9092
testtopics
Kafka 集群任意节点生产者测试
[root@server6 kraft]# /opt/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic testtopics
Kafka 集群任意节点查看主题
[root@server8 kraft]# /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic testtopics --from-beginning
Kafka 集群任意节点查看 topic 详细信息
[root@server6 kraft]# /opt/kafka/bin/kafka-topics.sh --bootstrap-server server7:9092,server8:9092,server6:9092 --topic testtopics --describe
FIlebeat 配置文件修改
使用:set paste可以再粘贴内容时不会自动缩进
配置kafka输出
注意将ES log 输出都注释掉
[root@server5 filebeat]# systemctl stop filebeat.service
检查语法
[root@server5 filebeat]# filebeat test config -c /etc/filebeat/filebeat.yml
Config OK
[root@server5 filebeat]# filebeat test output -c /etc/filebeat/filebeat.yml
启动
[root@server5 filebeat]# systemctl start filebeat.service
Kafka 集群任意节点消费者测试
[root@server7 kraft]# /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic apache-logs --from-beginning
Logstash 消费 Topic 消息
[root@server5 conf.d]#vimkafka.conf
[root@server5 conf.d]# /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/kafka.conf
启动后，观察控制台是否有报错。如果没有报错，说明 Logstash 已经成功连接上了 Kafka 和 ES。此时可以去 Kibana 中查看索引是否生成。