我要评论本篇文章继续给大家介绍elfk日志分析，我们先前介绍了elfk架构，zookeeper部署使用，kafka的部署，仅差kafka使用就将整个体系融汇贯通了。我们本篇文章将以kafka为核心，详细介绍kafka使用，最终将kafka融入elfk架构中，大致内容见下面目录。
目录
kafka集群原理
kafka学习使用中涉及许多原理，了解这些原理会让学习事半功倍。
一、专业术语
kafka cluster是分布式消息传递系统，与mq cluster（消息队列）类似，由broker list（kafka运行的节点）和多个topic（主题）组成，在kafka中，每个topic被细分为多个partition，而每个partition又可以被副本到多个kafka broker上实现高可用性。因此，kafka cluster是由多个broker节点和多个topic partition组成的。
producer是生产者角色，主要负责生产数据，是向kafka cluster写入数据的一方，数据的写入有两种常见策略，要么是rr算法，要么是基于key的hash值和分区数取余。
consumer是消费者角色，主要是负责消费数据，是从kafka cluster拉取数据的一方。
topic是主题，是数据存储的逻辑单元。
replica是副本，实际存储数据的地方，一个topic最少要有一个副本。
partition是分区，一个topic最少要有一个分区，正常情况下有多个分区编号的。副本是分区的实际载体。
consumer group是消费者组，一个消费者组里面最少有一个消费者，同一个消费者组的消费者不能同时消费同一个topic的partition，以免造成数据重复消费；当一个消费者组的消费者数据发生变化时，会触发rebalance（重平衡）机制，即重新分配分区消费。
isr是和leader数据同步的所有副本集合
osr是和leader数据不同步的所有副本集合
ar是isr和osr的集合，就是所有的副本集合
二、为什么kafka会丢数据
假如30秒内leader和follower数据的leo（log end offset）一致，则认为数据一致，当follower的数据还没有与leader完全同步时，leader节点宕机了，此时，follower选举出新的leader，其他的follower会跟随这个leader数据继续工作，如果之前leader恢复了，会从之前的hw（高水位线，isr中最后一个副本最小的leo）开始重新写数据，与新的leader同步，之前follower没同步的就丢失了，若30秒内，副本没有和leader的leo相同，会直接踢出isr，进入osr列表。
kafka集群基本使用
一、启动kafka
1、先启动zookeeper
[root@elk103 ~]# manager_zk.sh start
启动服务
========== elk101 zkserver.sh start ================
starting zookeeper ... started
========== elk102 zkserver.sh start ================
starting zookeeper ... started
========== elk103 zkserver.sh start ================
starting zookeeper ... started
[root@elk103 ~]# manager_zk.sh status
查看状态
========== elk101 zkserver.sh status ================
client port found: 2181. client address: localhost. client ssl: false.
mode: follower
========== elk102 zkserver.sh status ================
client port found: 2181. client address: localhost. client ssl: false.
mode: leader
========== elk103 zkserver.sh status ================
client port found: 2181. client address: localhost. client ssl: false.
mode: follower
2、模仿zookeeper启动脚本写一个kafka启动脚本，并给予执行权限
[root@elk101 ~]# cat /usr/local/sbin/manager_kafka.sh
#!/bin/bash
#判断用户是否传参
if [ $# -ne 1 ];then
echo "无效参数，用法为: $0 {start|stop|restart|status}"
exit
fi
#获取用户输入的命令
cmd=$1
#定义函数功能
function kafkamanger(){
case $cmd in
start)
echo "启动服务"
remoteexecution start
;;
stop)
echo "停止服务"
stopkafka stop
;;
restart)
echo "重启服务"
remoteexecution restart
;;
status)
echo "查看状态"
remoteexecution status
;;
*)
echo "无效参数，用法为: $0 {start|stop|restart|status}"
;;
esac
}
#定义执行的命令
function remoteexecution(){
for (( i=101 ; i<=103 ; i++ )) ; do
tput setaf 2
echo ========== elk${i} kafaka-server-start.sh $1 ================
tput setaf 9
ssh elk${i} "kafka-server-start.sh -daemon $kafka_home/config/server.properties"
done
}
function stopkafka(){
for (( i=101 ; i<=103 ; i++ )) ; do
tput setaf 2
echo ========== elk${i} kafaka-server-stop.sh $1 ================
tput setaf 9
ssh elk${i} "kafka-server-stop.sh -daemon $kafka_home/config/server.properties"
done
}
#调用函数
kafkamanger
[root@elk101 ~]# chmod +x /usr/local/sbin/manager_kafka.sh
[root@elk101 ~]# ll /usr/local/sbin/manager_kafka.sh
-rwxr-xr-x 1 root root 1323 jun 5 11:29 /usr/local/sbin/manager_kafka.sh
3、通过启动脚本启动kafka
[root@elk101 ~]# manager_kafka.sh start
启动服务
========== elk101 kafaka-server-start.sh start ================
========== elk102 kafaka-server-start.sh start ================
========== elk103 kafaka-server-start.sh start ================
二、topic管理
1、增
[root@elk101 ~]# kafka-topics.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --create --topic koten
created topic koten. #创建一个名为koten的topic
[root@elk101 ~]# kafka-topics.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --create --topic koten-3 --partitions 3
created topic koten-3. #创建一个名为koten-3的topic，分区数为3
[root@elk101 ~]# kafka-topics.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --create --topic koten-3-2 --partitions 3 --replication-factor 2
created topic koten-3-2. #创建一个名为koten-3-2的topic，分区数为3，副本数为2 2、查
#查看topic列表
[root@elk101 ~]# kafka-topics.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --list
koten
koten-3
koten-3-2
#查看所有topic的详细信息
[root@elk101 ~]# kafka-topics.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --describe
topic: koten-3-2 topicid: 1l4p-tv_q3ataskediu3mq partitioncount: 3 replicationfactor: 2 configs: segment.bytes=1073741824
topic: koten-3-2 partition: 0 leader: 103 replicas: 103,102 isr: 103
topic: koten-3-2 partition: 1 leader: 102 replicas: 102,101 isr: 102,101
topic: koten-3-2 partition: 2 leader: 101 replicas: 101,103 isr: 101,103
topic: koten topicid: exxgjwbysxe_wax-fv83za partitioncount: 1 replicationfactor: 1 configs: segment.bytes=1073741824
topic: koten partition: 0 leader: 103 replicas: 103 isr: 103
topic: koten-3 topicid: l7l3sy63qv-ayxod46bviq partitioncount: 3 replicationfactor: 1 configs: segment.bytes=1073741824
topic: koten-3 partition: 0 leader: 103 replicas: 103 isr: 103
topic: koten-3 partition: 1 leader: 102 replicas: 102 isr: 102
topic: koten-3 partition: 2 leader: 101 replicas: 101 isr: 101
#查看指定topic的详细信息
oot@elk101 ~]# kafka-topics.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --describe --topic koten
topic: koten topicid: exxgjwbysxe_wax-fv83za partitioncount: 1 replicationfactor: 1 configs: segment.bytes=1073741824
topic: koten partition: 0 leader: 103 replicas: 103 isr: 1033、改
主分片数可以用命令行直接修改，副本数修改比较麻烦，需要用json格式，可以参考这个大神的链接：https://www.cnblogs.com/yinzhengjie/p/9808125.html
#修改koten的topic分区为5个
[root@elk101 ~]# kafka-topics.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --alter --topic koten --partitions 54、删
[root@elk101 ~]# kafka-topics.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --delete --topic koten三、生产者和消费者
1、启动生产者
[root@elk101 ~]# kafka-console-producer.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --topic koten-32、启动消费者（在同一主机）
表示从最新的offset拉取数据
[root@elk101 ~]# kafka-console-consumer.sh --bootstrap-server 10.0.0.102:9092 --topic koten-3
表示从头开始拉取数据
[root@elk101 ~]# kafka-console-consumer.sh --bootstrap-server 10.0.0.102:9092 --topic koten-3 --from-beginning四、消费者组管理
1、查看现有的消费者组
[root@elk101 ~]# kafka-consumer-groups.sh --bootstraerver 10.0.0.101:9092 --list
console-consumer-24702
console-consumer-58734
console-consumer-41114
2、列出所有消费者组的详细信息，包括偏移量，消费者id，leo等信息
[root@elk101 ~]# kafka-consumer-groups.sh --bootstrap-server 10.0.0.101:9092 --describe --all-groups
consumer group 'console-consumer-24702' has no active members.
consumer group 'console-consumer-41114' has no active members.
group topic partition current-offset log-end-offset lag consumer-id host client-id
console-consumer-58734 koten-3 0 - 1 - console-consumer-f92294eb-383c-402b-9f9e-9a7ac5773b7d /10.0.0.101 console-consumer
console-consumer-58734 koten-3 1 - 2 - console-consumer-f92294eb-383c-402b-9f9e-9a7ac5773b7d /10.0.0.101 console-consumer
console-consumer-58734 koten-3 2 - 3 - console-consumer-f92294eb-383c-402b-9f9e-9a7ac5773b7d /10.0.0.101 console-consumer
3、查看内置topic的offset数据，了解即可，我这边运行没有显示内容
[root@elk101 ~]# kafka-console-consumer.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --topic __consumer_offsets --formatter "kafka.coordinator.group.groupmetadatamanager\$offsetsmessageformatter" --from-beginning4、基于配置文件指定消费组
[root@elk101 ~]# cat $kafka_home/config/consumer.properties
......
group.id=koten-consumer-group
......
[root@elk101 ~]# kafka-console-consumer.sh --bootstrap-server 10.0.0.101:9092 --topic koten-topic --consumer.config $kafka_home/config/consumer.properties
[2023-06-05 20:25:33,829] warn [consumer clientid=console-consumer, groupid=koten-consumer-group] error while fetching metadata with correlation id 2 : {koten-topic=leader_not_available} (org.apache.kafka.clients.networkclient)5、基于命令行参数指定消费者组
[root@elk101 ~]# kafka-console-consumer.sh --bootstrap-server 10.0.0.101:9092 --topic koten-topic --consumer-property group.id=koten-consumer-groupzookeeper堆内存调优
生产环境建议调到2g~4g
1、查zookeeper的堆内存大小
[root@elk101 ~]# jmap -heap `jps | awk '/quorumpeermain/{print $1}'` | grep maxheapsize
maxheapsize = 1048576000 (1000.0mb)
[root@elk101 ~]# 2、修改堆内存的大小
[root@elk101 ~]# cat > /koten/softwares/apache-zookeeper-3.8.1-bin/conf/java.env <<'eof'
#!/bin/bash
# 指定jdk的按住路径
export java_home=/koten/softwares/jdk1.8.0_291
# 指定zookeeper的堆内存大小
export jvmflags="-xms128m -xmx128m $jvmflags"
eof
3、将配置文件同步到集群的其他zk节点上
[root@elk101 ~]# data_rsync.sh /koten/softwares/apache-zookeeper-3.8.1-bin/conf/java.env
===== rsyncing elk102: java.env =====
命令执行成功!
===== rsyncing elk103: java.env =====
命令执行成功!
4、重启zk集群，注意一定要重启后堆内存才生效，manager_zk脚本不好用，有时候停止了，进程还存在，需要手动挨个运行zkserver.sh stop
[root@elk101 ~]# manager_zk.sh restart
重启服务
========== elk101 zkserver.sh restart ================
stopping zookeeper ... stopped
starting zookeeper ... started
========== elk102 zkserver.sh restart ================
stopping zookeeper ... stopped
starting zookeeper ... started
========== elk103 zkserver.sh restart ================
stopping zookeeper ... stopped
starting zookeeper ... started5、验证堆内存
[root@elk102 ~]# jmap -heap `jps | awk '/quorumpeermain/{print $1}'` | grep maxheapsize
maxheapsize = 268435456 (256.0mb)
kafka堆内存调优
1、查看堆内存大小
[root@elk101 ~]# jmap -heap `jps | awk '/kafka/{print $1}'` | grep maxheapsize
maxheapsize = 1073741824 (1024.0mb)2、修改堆内存（生产环境5~6g最佳）
捎带启动了jxm端口
[root@elk101 ~]# cat `which kafka-server-start.sh`
......
if [ "x$kafka_heap_opts" = "x" ]; then
# export kafka_heap_opts="-xmx1g -xms1g"
export kafka_heap_opts="-server -xmx256m -xms256m -
xx:permsize=128m -xx:+useg1gc -xx:maxgcpausemillis=200
-xx:parallelgcthreads=8 -xx:concgcthreads=5 -xx:initiat
ingheapoccupancypercent=70"
export jmx_port="8888"
fi
......3、单点重启kafka，查看堆内存大小
[root@elk101 ~]# kafka-server-stop.sh
[root@elk101 ~]# kafka-server-start.sh -daemon $kafka_home/config/server.properties
[root@elk101 ~]# jmap -heap `jps | awk '/kafka/{print $1}'` | grep maxheapsize
maxheapsize = 268435456 (256.0mb)4、将启动脚本同步到其他节点
[root@elk101 ~]# data_rsync.sh `which kafka-server-start.sh`
===== rsyncing elk102: kafka-server-start.sh =====
命令执行成功!
===== rsyncing elk103: kafka-server-start.sh =====
命令执行成功!
5、其他节点重启kafka环境，查看堆内存是否生效
[root@elk101 ~]# manager_kafka.sh stop
停止服务
========== elk101 kafaka-server-stop.sh stop ================
========== elk102 kafaka-server-stop.sh stop ================
========== elk103 kafaka-server-stop.sh stop ================
[root@elk101 ~]# manager_kafka.sh start
启动服务
========== elk101 kafaka-server-start.sh start ================
========== elk102 kafaka-server-start.sh start ================
========== elk103 kafaka-server-start.sh start ================
[root@elk102 ~]# jmap -heap `jps | awk '/kafka/{print $1}'` | grep maxheapsize
maxheapsize = 268435456 (256.0mb)
[root@elk103 ~]# jmap -heap `jps | awk '/kafka/{print $1}'` | grep maxheapsize
maxheapsize = 268435456 (256.0mb)
kafka开源监控组件kafka-eagle
图形化的方式管理kafka
一、前期准备
1、启动kafka的jmx端口
与上面修改堆内存步骤相同，略。
2、启动zookeeper的jmx端口
[root@elk101 ~]# cat /koten/softwares/apache-zookeeper-3.8.1-bin/bin/zkenv.sh
# zookeeper jmx
jmxlocalonly=false
jmxhostname=10.0.0.101
jmxport=9999
jmxssl=false
jmxlog4j=false3、安装mysql，启动服务并设置开机自启动
[root@elk101 ~]# yum -y install mariadb-server
[root@elk101 ~]# cat /etc/my.cnf
[mysqld]
......
skip-name-resolve=1 #跳过名称解析，不跳过再进行登录的时候，可能会进行反向解析
[root@elk101 ~]# systemctl enable --now mariadb
created symlink from /etc/systemd/system/multi-user.target.wants/mariadb.service to /usr/lib/systemd/system/mariadb.service.
4、创建数据库与授权用户
[root@elk101 ~]# mysql
welcome to the mariadb monitor. commands end with ; or \g.
your mariadb connection id is 2
server version: 5.5.68-mariadb mariadb server
copyright (c) 2000, 2018, oracle, mariadb corporation ab and others.
type 'help;' or '\h' for help. type '\c' to clear the current input statement.
mariadb [(none)]> create database koten_kafka default character set utf8mb4 collate utf8mb4_unicode_ci;
query ok, 1 row affected (0.00 sec)
mariadb [(none)]> create user admin identified by 'koten';
query ok, 0 rows affected (0.00 sec)
mariadb [(none)]> grant all on koten_kafka.* to admin;
query ok, 0 rows affected (0.00 sec)
mariadb [(none)]> show grants for admin;
+------------------------------------------------------------------------------------------------------+
| grants for admin@% |
+------------------------------------------------------------------------------------------------------+
| grant usage on *.* to 'admin'@'%' identified by password '*87f5f6ff9376d7c33feb4c2aa1f7f99e9853f2db' |
| grant all privileges on `koten_kafka`.* to 'admin'@'%' |
+------------------------------------------------------------------------------------------------------+
2 rows in set (0.00 sec)
mariadb [(none)]> quit
bye5、测试用户
[root@elk101 ~]# mysql -u admin -pkoten -h 10.0.0.101
welcome to the mariadb monitor. commands end with ; or \g.
your mariadb connection id is 3
server version: 5.5.68-mariadb mariadb server
copyright (c) 2000, 2018, oracle, mariadb corporation ab and others.
type 'help;' or '\h' for help. type '\c' to clear the current input statement.
mariadb [(none)]> show databases;
+--------------------+
| database |
+--------------------+
| information_schema |
| koten_kafka |
| test |
+--------------------+
3 rows in set (0.00 sec)
mariadb [(none)]> quit
bye
二、部署监控
1、下载kafka-eagle软件，下面的链接下载的慢可以用我分享在文末的链接
[root@elk101 ~]# wget https://github.com/smartloli/kafka-eagle-bin/archive/v2.0.8.tar.gz2、解压软件包
[root@elk101 ~]# unzip kafka-eagle-bin-2.0.8.zip
archive: kafka-eagle-bin-2.0.8.zip
inflating: efak-web-2.0.8-bin.tar.gz
inflating: system-config.properties
[root@elk101 ~]# tar xf efak-web-2.0.8-bin.tar.gz -c /koten/softwares/
3、修改配置文件
[root@elk101 ~]# cat /koten/softwares/efak-web-2.0.8/conf/system-config.properties
efak.zk.cluster.alias=kafka
kafka.zk.list=10.0.0.101:2181,10.0.0.102:2181,10.0.0.103:2181/kafka-3.2.1 #注意chroot与kafka配置文件保持一致
kafka.efak.broker.size=20
kafka.zk.limit.size=16
efak.webui.port=8048
kafka.efak.offset.storage=zk
kafka.efak.jmx.uri=service:jmx:rmi:///jndi/rmi://%s/jmxrmi
efak.metrics.charts=true
efak.metrics.retain=15
efak.sql.topic.records.max=5000
efak.sql.topic.preview.records.max=10
efak.topic.token=koten
efak.driver=com.mysql.cj.jdbc.driver
efak.url=jdbc:mysql://10.0.0.101:3306/koten_kafka?useunicode=true&characterencodi
ng=utf-8&zerodatetimebehavior=converttonull
efak.username=admin
efak.password=koten #数据库密码4、配置环境变量
[root@elk101 ~]# cat >> /etc/profile.d/kafka.sh <<'eof'
export ke_home=/koten/softwares/efak-web-2.0.8
export path=$path:$ke_home/bin
eof
[root@elk101 ~]# source /etc/profile.d/kafka.sh
5、修改配置启动脚本的堆内存大小
[root@elk101 ~]# sed -i '/ke_java_opts/s#2g#256m#g' $ke_home/bin/ke.sh | grep ke_java_opts
[root@elk101 ~]# grep ke_java_opts $ke_home/bin/ke.sh
export ke_java_opts="-server -xmx256m -xms256m -xx:maxgcpausemillis=20 -xx:+useg1gc -xx:metaspacesize=128m -xx:initiatingheapoccupancypercent=35 -xx:g1heapregionsize=16m -xx:minmetaspacefreeratio=50 -xx:maxmetaspacefreeratio=80"
6、启动服务
[root@elk101 ~]# ke.sh start
......
[2023-06-05 22:36:46] info: [job done!]
welcome to
______ ______ ___ __ __
/ ____/ / ____/ / | / //_/
/ __/ / /_ / /| | / ,<
/ /___ / __/ / ___ | / /| |
/_____/ /_/ /_/ |_|/_/ |_|
( eagle for apache kafka® )
version 2.0.8 -- copyright 2016-2021
*******************************************************************
* efak service has started success.
* welcome, now you can visit 'http://10.0.0.101:8048'
* account:admin ,password:123456
*******************************************************************
* <usage> ke.sh [start|status|stop|restart|stats] </usage>
* <usage> https://www.kafka-eagle.org/ </usage>
*******************************************************************
7、登录eagle
8、忘记密码后可以进入数据库去找
mariadb [koten_kafka]> select * from koten_kafka.ke_users;
+----+-------+----------+-----------+-----------------+---------------+
| id | rtxno | username | password | email | realname |
+----+-------+----------+-----------+-----------------+---------------+
| 1 | 1000 | admin | 123456 | admin@email.com | administrator |
+----+-------+----------+-----------+-----------------+---------------+
1 row in set (0.00 sec)
mariadb [koten_kafka]> 9、登陆进去后可以看到监控的kafka与zookeeper的一些数据信息
查看仪表盘
查看zookeeper与kafka的监控信息
10、不止可以监控数据，还可以创建topic，对kafka集群进行一些操作
kafka集群压力测试
对kafka集群进行压力测试，方便我们了解集群的处理上限，可以作为集群调优和扩容的依据。
测试之前要先搞懂链路，如果你的生产者与kafka集群不在一个地方，那么你在一个主机进行压力测试也没有意义，确定好链路后，在实际的生产者和消费者执行压测，修改对应的主机参数即可。
1、通过下面脚本进行压测
mkdir /tmp/kafka-test
cat > kafka-test.sh <<'eof'
# 创建topic
kafka-topics.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --topic kafka-2023 --replication-factor 1 --partitions 10 --create
# 启动消费者消费数据
nohup kafka-consumer-perf-test.sh --broker-list 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --topic kafka-2023 --messages 100000 &>/tmp/kafka-test/kafka-consumer.log &
# 启动生产者写入数据
nohup kafka-producer-perf-test.sh --num-records 100000 --record-size 1000 --topic kafka-2023 --throughput 1000 --producer-props bootstrap.servers=10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 &> /tmp/kafka-test/kafka-producer.log &
eof
bash kafka-test.sh
#可以根据自己要测的实际生产情况调整以下参数
kafka-consumer-perf-test.sh
---messages: 指定消费消息的数量。
--broker-list: 指定broker列表。
--topic: 指定topic主体。
kafka-producer-perf-test.sh
-num-records：生产消息的数量。
--record-size: 每条消息的大小，单位是字节。
--topic: 指定topic主体。
--throughput: 设置每秒发送的消息数量，即指定最大消息的吞吐量，若设置为-1代表不限制！
--producer-props bootstrap.servers: 指定broker列表2、可以通过efak查看实施进度
3、也可以通过脚本输出的日志观察生产和消费速度
filebeat对接kafka
filebeat对接kafka，如果filebeat作为生产者，kafka作为消费者，可以经过kafka后再写入到es集群；filebeat作为消费者，可以读取到kafka的数据，将kafka的数据展示出来，或者写入es集群。下面给大家展示下示例。
一、filebeat作为生产者
filebeat作为生产者，所以是output到kafka，input我们就用stdin去测试。
1、kafka启动消费者
[root@elk101 ~]# kafka-console-consumer.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --topic filebeat-to-kafka2、执行filebeat
[root@elk101 config]# cat 01-stdin-to-kafka.yaml
filebeat.inputs:
- type: stdin
output.kafka:
hosts: ["10.0.0.101:9092", "10.0.0.102:9092", "10.0.0.103:9092"]
topic: 'filebeat-to-kafka'
# 执行filebeat并输入测试数据
[root@elk101 config]# filebeat -e -c config/01-stdin-to-kafka.yaml
...
1234567
# kafka会消费到数据
[root@elk101 ~]# kafka-console-consumer.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --topic filebeat-to-kafka
[2023-12-18 15:51:33,774] warn [consumer clientid=console-consumer, groupid=console-consumer-60747] error while fetching metadata with correlation id 2 : {filebeat-to-kafka=leader_not_available} (org.apache.kafka.clients.networkclient)
{"@timestamp":"2023-12-18t07:54:06.389z","@metadata":{"beat":"filebeat","type":"_doc","version":"7.17.5"},"ecs":{"version":"1.12.0"},"log":{"offset":0,"file":{"path":""}},"message":"1234567","input":{"type":"stdin"},"host":{"name":"elk101"},"agent":{"id":"8fa0a9d7-f6d8-45b8-9355-ddf800e337fa","name":"elk101","type":"filebeat","version":"7.17.5","hostname":"elk101","ephemeral_id":"cef5e36c-ef9b-4c38-91d9-54f7c4db48fe"}}二、filebeat作为消费者
1、启动filebeat进行消费
[root@elk101 config]# cat 02-kafka-to-filebeat.yaml
filebeat.inputs:
- type: kafka
hosts:
- 10.0.0.101:9092
- 10.0.0.102:9092
- 10.0.0.103:9092
topics: ["kafka-to-filebeat"]
group_id: "filebeat"
output.console:
pretty: true
[root@elk101 config]# filebeat -e -c config/02-kafka-to-filebeat.yaml2、启动kafka进行生产
[root@elk101 ~]# kafka-console-producer.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --topic kafka-to-filebeat
>123456ceshi
>
[root@elk101 config]# filebeat -e -c config/02-kafka-to-filebeat.yaml
...
{
"@timestamp": "2023-12-18t09:06:52.226z",
"@metadata": {
"beat": "filebeat",
"type": "_doc",
"version": "7.17.5"
},
"ecs": {
"version": "1.12.0"
},
"host": {
"name": "elk101"
},
"agent": {
"version": "7.17.5",
"hostname": "elk101",
"ephemeral_id": "d4c29b42-d892-4532-bbe6-cff5f5a243f9",
"id": "8fa0a9d7-f6d8-45b8-9355-ddf800e337fa",
"name": "elk101",
"type": "filebeat"
},
"kafka": {
"partition": 0,
"offset": 0,
"key": "",
"headers": [],
"topic": "kafka-to-filebeat"
},
"message": "123456ceshi",
"input": {
"type": "kafka"
}
}
logstash对接kafka
一、logstash作为生产者
1、kafaka开始消费
[root@elk101 ~]# kafka-console-consumer.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --topic logstash-to-kafka2、启动logstash开始生产
[root@elk101 logstash_cofig]# cat 01-logstash-to-kafka.yaml
input {
stdin {
}
}
output {
kafka {
bootstrap_servers => "10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092"
topic_id => "logstash-to-kafka"
}
}
# 启动logstash并手动写入数据
[root@elk101 logstash_cofig]# logstash -r -f 01-logstash-to-kafka.yaml
...
ceshi123
# kafka可以消费到数据
[root@elk101 ~]# kafka-console-consumer.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --topic logstash-to-kafka
...
2023-12-18t08:04:25.969z elk101
2023-12-18t08:05:03.169z elk101 ceshi123二、logstash作为消费者
1、启动logstash进行消费
[root@elk101 logstash_cofig]# cat 02-kafka-to-logstasg.yaml
input {
kafka {
bootstrap_servers => "10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092"
topics => "kafka-to-logstash"
}
}
output {
stdout {}
}
[root@elk101 logstash_cofig]# logstash -r -f 02-kafka-to-logstasg.yaml2、启动kafka进行生产
[root@elk101 ~]# kafka-console-producer.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --topic kafka-to-logstash
>123
>
[root@elk101 logstash_cofig]# logstash -r -f 02-kafka-to-logstasg.yaml
...
{
"@version" => "1",
"@timestamp" => 2023-12-18t09:19:51.170z,
"message" => "123"
}
kafka图形化管理工具下载链接：https://pan.baidu.com/s/1_xcim_6oc0383f11phrycq?pwd=j7ki
我是koten，10年运维经验，持续分享运维干货，感谢大家的阅读和关注！
发表评论