kafka filebeat输出到kafka Logstash 消费 Topic 消息

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