Kafka 集群部署与 ELFK 日志集成指南

ELFK kafka 架构部署内存紧张修改sever1、serever2、server3配置[rootserver1 ~]# cd /etc/elasticsearch/[rootserver1 elasticsearch]# vim elasticsearch.ymlbootstrap.memory_lock: false[rootserver1 elasticsearch]# vim jvm.options-Xms512m-Xmx512m[rootserver1 elasticsearch]# systemctl restart elasticsearchserver2、server3做同样配置集群部署再开启三台虚拟机server6、server7、server8基础配置互相解析三台主机安装 JDK 环境浏览器下载jdk-17.0.17_linux-x64_bin.rpm[rootserver6 ~]# rpm -ivh jdk-17.0.17_linux-x64_bin.rpm[rootserver6 ~]# java -version三台主机安装 Kafka浏览器下载kafka_2.12-3.9.1.tgzhttps://kafka.apache.org/downloads[rootserver6 ~]# tar zxf kafka_2.12-3.9.1.tgz -C /opt[rootserver6 ~]# mv /opt/kafka_2.12-3.9.1 /opt/kafka[rootserver6 ~]# cd /opt/kafka/config/kraft[rootserver6 kraft]# cp server.properties{,.bak}备份配置文件[rootserver6 kraft]# mkdir /opt/kafka/logs 创建日志目录三台主机 Kafka 配置文件修改[rootserver6 kraft]# vim server.propertiesprocess.rolesbroker,controllernode.id176controller.quorum.voters176192.168.164.176:9093,177192.168.164.177:9093,178192.168.164.178:9093listenersPLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093inter.broker.listener.namePLAINTEXTadvertised.listenersPLAINTEXT://192.168.164.176:9092,CONTROLLER://192.168.164.176:9093controller.listener.namesCONTROLLERlistener.security.protocol.mapCONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSLnum.network.threads2num.io.threads2socket.send.buffer.bytes102400socket.receive.buffer.bytes102400socket.request.max.bytes104857600log.dirs/opt/kafka/logsnum.partitions3num.recovery.threads.per.data.dir1offsets.topic.replication.factor3transaction.state.log.replication.factor3transaction.state.log.min.isr1log.retention.hours168log.segment.bytes1073741824log.retention.check.interval.ms300000注意server2、server3配置区别在于node.id、advertised.listenersKafka 集群初始化server6 节点生成储目录唯一的 UUID[rootserver6 kraft]# KAFKA_CLUSTER_ID$(/opt/kafka/bin/kafka-storage.sh random-uuid)[rootserver6 kraft]# echo ${KAFKA_CLUSTER_ID}K9eaFC2PSDSKruBsoJ-oTw三台主机用该 uuid 格式化 kafka 存储目录注意uuid[rootserver6 kraft]# /opt/kafka/bin/kafka-storage.sh format -t K9eaFC2PSDSKruBsoJ-oTw -c /opt/kafka/config/kraft/server.properties[rootserver7 kraft]# /opt/kafka/bin/kafka-storage.sh format -t K9eaFC2PSDSKruBsoJ-oTw -c /opt/kafka/config/kraft/server.properties[rootserver8 kraft]# /opt/kafka/bin/kafka-storage.sh format -t K9eaFC2PSDSKruBsoJ-oTw -c /opt/kafka/config/kraft/server.properties三台节点启动 Kafkaserver6上[rootserver6 kraft]# /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/kraft/server.properties[rootserver6 kraft]# jps4261 Kafka4332 Jpsserver7上[rootserver7 kraft]# /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/kraft/server.properties[rootserver7 kraft]# jps6039 Kafka6109 Jpsserver8上[rootserver8 kraft]# /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/kraft/server.properties[rootserver8 kraft]# jps3667 Kafka3688 JpsKafka 集群可用性验证Kafka 集群任意节点创建 Topic[rootserver6 kraft]# /opt/kafka/bin/kafka-topics.sh --create --topic testtopics --bootstrap-server localhost:9092 --replication-factor 3 --partitions 3Created topic testtopics.--replication-factor 3设置主题的副本因子为 3即每个分区的副本数量为 3。副本用于数据冗余和高可用性副本数需不能超过集群中 Broker 的总数。--partitions 3设置主题的分区数为 3。分区是 Kafka 并行处理消息的基本单位影响吞吐量和扩展性。分区数一旦创建不可减少。Kafka 集群任意节点查看 Topic[rootserver7 kraft]# /opt/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:9092testtopicsKafka 集群任意节点生产者测试[rootserver8 kraft]# /opt/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic testtopics111222333Kafka 集群任意节点查看主题[rootserver7 kraft]# /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic testtopics --from-beginning111222333Processed a total of 3 messagesKafka 集群任意节点查看 topic 详细信息[rootserver6 kraft]# /opt/kafka/bin/kafka-topics.sh --bootstrap-server server7:9092,server8:9092,server9:9092 --topic testtopics --describeFIlebeat 配置文件修改server4上[rootserver4 ~]# vim /etc/filebeat/filebeat.yml 注释掉logstash部分#----------------------------- Kafka output --------------------------------output.kafka:hosts: [192.168.164.176:9092, 192.168.164.177:9092, 192.168.164.178:9092]topic: apache-logspartition.round_robin:reachable_only: truerequired_acks: 1compression: gzip[rootserver4 ~]# filebeat test config -c /etc/filebeat/filebeat.ymlConfig OK[rootserver4 ~]# filebeat test output -c /etc/filebeat/filebeat.yml[rootserver4 ~]# systemctl restart filebeat.serviceKafka 集群任意节点消费者测试[rootserver6 ~]# /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic apache-logs --from-beginningLogstash 消费 Topic 消息[rootserver4 modules.d]# cd[rootserver5 conf.d]# cd /etc/logstash/conf.d[rootserver5 conf.d]# vim kafka.confinput {kafka {bootstrap_servers 192.168.164.176:9092,192.168.164.177:9092,192.168.164.178:9092topics [apache-logs]group_id logstash-consumer-groupconsumer_threads 3auto_offset_reset earliestcodec json}}filter {grok {match { message %{HTTPD_COMBINEDLOG} }}}output {#stdout {}elasticsearch {hosts [192.168.164.171:9200,192.168.164.172:9200,192.168.164.173:9200]index apachelog-%{YYYY.MM.dd}user elasticpassword westos}}[rootserver5 conf.d]# /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/kafka.conf