Zookeeper、Kafka集群部署（Filebeat+ELK环境）

ELK+filebeat环境可以参考前面的博客，本次环境基于之前的配置

Server	IP	Components	System
k8s-master01	192.168.126.21	Logstash、Apache、Zookeeper、Kafka	CentOS7.4(64 位)
k8s-worker01	192.168.126.22	Filebeat、ElasticSearch 、Zookeeper、Kafka	CentOS7.4(64 位)
k8s-worker02	192.168.126.23	ElasticSearch、kibana、Zookeeper、Kafka	CentOS7.4(64 位)

Zookeeper

Zookeeper简介

ZooKeeper是一个分布式的，开放源码的分布式应用程序协调服务，是Google的Chubby一个开源的实现，是Hadoop和Hbase的重要组件。它是一个为分布式应用提供一致性服务的软件，提供的功能包括：配置维护、域名服务、分布式同步、组服务等。

zookeeper用于协调Kafka的分布式消息系统。

Zookeeper下载安装

官方下载地址 https://archive.apache.org/dist/zookeeper/

wget https://archive.apache.org/dist/zookeeper/zookeeper-3.6.4/apache-zookeeper-3.6.4-bin.tar.gz

速度较慢可以先下载后上传到主机
tar -zxf  apache-zookeeper-3.6.4-bin.tar.gz -C /usr/local/
ln -s /usr/local/apache-zookeeper-3.6.4-bin/ /usr/local/zookeeper
cd  /usr/local/zookeeper/conf/
mkdir /usr/local/zookeeper/data
mkdir /usr/local/zookeeper/logs

获取配置文件
cp zoo_sample.cfg zoo.cfg
vim zoo.cfg

[root@k8s-master01 conf]# cat zoo.cfg | grep -Ev '^(#|$)'
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/usr/local/zookeeper/data #修改数据位置
dataLogDir=/usr/local/zookeeper/logs #添加日志位置
clientPort=2181
server.1=192.168.126.21:3188:3288 #添加服务
server.2=192.168.126.22:3188:3288
server.3=192.168.126.23:3188:3288

复制配置到其他主机
scp zoo.cfg root@192.168.126.22:/usr/local/zookeeper/conf/

server.A=B:C:D
-A是一个数字，表示这个是第几号服务器。集群模式下需要在zoo.cfg中dataDir指定的目录下创建一个文件myid，这个文件里面有一个数据就是A的值，Zookeeper启动时读取此文件，拿到里面的数据与zoo.cfg里面的配置信息比较从而判断到底是哪个server。
-B是这个服务器的地址。
-C是这个服务器Follower与集群中的Leader服务器交换信息的端口。
-D是万一集群中的Leader服务器挂了，需要一个端口来重新进行选举，选出一个新的Leader，而这个端口就是用来执行选举时服务器相互通信的端口。

给zookeeper配置myid
echo 1 > /usr/local/zookeeper/data/myid
echo 2 > /usr/local/zookeeper/data/myid
echo 3 > /usr/local/zookeeper/data/myid

cat > /etc/init.d/zookeeper <<'EOF'
#!/bin/bash
#chkconfig: 2345 20 90
#description:Zookeeper Service Control Script
ZK_HOME='/usr/local/zookeeper'
case $1 in
start)
    echo "---------- zookeeper 启动 ------------"
    $ZK_HOME/bin/zkServer.sh start
;;
stop)
    echo "---------- zookeeper 停止 ------------"
    $ZK_HOME/bin/zkServer.sh stop
;;
restart)
    echo "---------- zookeeper 重启 ------------"
    $ZK_HOME/bin/zkServer.sh restart
;;
status)
    echo "---------- zookeeper 状态 ------------"
    $ZK_HOME/bin/zkServer.sh status
;;
*)
    echo "Usage: $0 {start|stop|restart|status}"
esac
EOF
chmod +x /etc/init.d/zookeeper
#启动 Zookeeper
service zookeeper start

#设置开机自启
chkconfig --add zookeeper

查看状态
service zookeeper status

Kafka

Kafka如果是3.0版本，由于元信息储存在本地，不需要zk，可以直接部署。

Kafka简介

Kafka是最初由Linkedin公司开发，是一个分布式、支持分区的（partition）、多副本的（replica），基于zookeeper协调的分布式消息系统。

Kafka可以实时的处理大量数据以满足各种需求场景：比如基于hadoop的批处理系统、低延迟的实时系统、storm/Spark流式处理引擎，web/nginx日志、访问日志，消息服务等等。

kafka有以下特性
1）高吞吐量、低延迟
Kafka 每秒可以处理几十万条消息，它的延迟最低只有几毫秒。对消息的生产和消费是异步的。

2）可扩展性
kafka 集群支持热扩展

3）持久性、可靠性
消息被持久化到本地磁盘，并且支持数据备份防止数据丢失

4）容错性
允许集群中节点失败（多副本情况下，若副本数量为 n，则允许 n-1 个节点失败）

5）并发性
支持数千个客户端同时读写，

kafka中一些词的基本概念

• Producer：Producer即生产者，消息的产生者，是消息的入口。
• Consumer：消费者，即消息的消费方，是消息的出口。
• Consumer Group：我们可以将多个消费组组成一个消费者组，在kafka的设计中同一个分区的数据只能被消费者组中的某一个消费者消费。同一个消费者组的消费者可以消费同一个topic的不同分区的数据，这也是为了提高kafka的吞吐量！
• Broker：Broker是kafka实例，每个服务器上有一个或多个kafka的实例，我们姑且认为每个broker对应一台服务器。每个kafka集群内的broker都有一个不重复的编号
• Topic：消息的主题，可以理解为消息的分类，kafka的数据就保存在topic。在每个broker上都可以创建多个topic。
• Partition：Topic的分区，每个topic可以有多个分区，分区的作用是做负载，提高kafka的吞吐量。同一个topic在不同的分区的数据是不重复的，partition的表现形式就是一个一个的文件夹！
• Replication:每一个分区都有多个副本，副本的作用是做备胎。当主分区（Leader）故障的时候会选择一个备胎（Follower）上位，成为Leader。在kafka中默认副本的最大数量是10个，且副本的数量不能大于Broker的数量，follower和- leader绝对是在不同的机器，同一机器对同一个分区也只可能存放一个副本（包括自己）。
• Leader： 每个 partition 有多个副本，其中有且仅有一个作为 Leader，Leader 是当前负责数据的读写的 partition。
• Follower：Follower 跟随 Leader，所有写请求都通过 Leader 路由，数据变更会广播给所有 Follower，Follower 与 Leader 保持数据同步。Follower 只负责备份，不负责数据的读写。
• Message：每一条发送的消息主体。
• offset：偏移量，消费者通过偏移量来决定下次读取的消息（即消费位置）。
  消息被消费之后，并不被马上删除，这样多个业务就可以重复使用 Kafka 的消息。
• Zookeeper：kafka集群依赖zookeeper来保存集群的的元信息，来保证系统的可用性。Kafka如果是3.0版本不需要。

kafka安装配置

官网下载地址 http://kafka.apache.org/downloads.html

wegt https://archive.apache.org/dist/kafka/2.8.0/kafka_2.13-2.8.0.tgz

tar -zxvf kafka_2.13-2.8.0.tgz -C /usr/local/
ln -s /usr/local/kafka_2.13-2.8.0/ /usr/local/kafka

cd /usr/local/kafka/config/
cp server.properties{,.bak}

#新建日志目录
mkdir /usr/local/kafka/logs

vim server.properties

[root@k8s-master01 config]# cat server.properties | grep -Ev '^(#|$)'
broker.id=2
listeners=PLAINTEXT://192.168.126.21:9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/usr/local/kafka/logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=192.168.126.21:2181,192.168.126.22:2181,192.168.126.23:2181
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0

broker.id=0 #id必须不一致
listeners=PLAINTEXT://192.168.126.21:9092  #修改为本机ip
num.network.threads=3                     #处理网络请求的线程数量
num.io.threads=8                          #用来处理磁盘IO的线程数量
socket.send.buffer.bytes=102400           #发送套接字的缓冲区大小
socket.receive.buffer.bytes=102400        #接受套接字的缓冲区大小
socket.request.max.bytes=104857600        #请求套接字的缓冲区大小
log.dirs=/usr/local/kafka/logs            #修改日志位置
num.partitions=1                          #topic的默认分区个数，会被topic创建时的指定参数覆盖
num.recovery.threads.per.data.dir=1        #恢复和清理data下数据的线程数量
offsets.topic.replication.factor=1        #表示kafka的内部topic consumer_offsets副本数
transaction.state.log.replication.factor=1 #事务主题的复制因子
transaction.state.log.min.isr=1           #覆盖事务主题的min.insync.replicas配置
log.retention.hours=168                   #日志保留的最小时间
log.segment.bytes=1073741824              #最大segment
log.retention.check.interval.ms=300000    #检查日志段文件的间隔时间，以确定是否文件属性是否到达删除要
zookeeper.connect=192.168.126.21:2181,192.168.126.22:2181,192.168.126.23:2181 #zookeeper地址 
zookeeper.connection.timeout.ms=18000     #超时时间
group.initial.rebalance.delay.ms=0        #coordinator推迟空消费组接收到成员加入请求后本应立即开启的rebalance的时间

cat > /etc/init.d/kafka<<'EOF'
#!/bin/bash
#chkconfig:2345 22 88
#description:Kafka Service Control Script
KAFKA_HOME='/usr/local/kafka'
case $1 in
start)
    echo "---------- Kafka 启动 ------------"
    ${KAFKA_HOME}/bin/kafka-server-start.sh -daemon ${KAFKA_HOME}/config/server.properties
;;
stop)
    echo "---------- Kafka 停止 ------------"
    ${KAFKA_HOME}/bin/kafka-server-stop.sh
;;
restart)
    $0 stop
    $0 start
;;
status)
    echo "---------- Kafka 状态 ------------"
    count=$(ps -ef | grep kafka | egrep -cv "grep|$$")
    if [ "$count" -eq 0 ];then
        echo "kafka is not running"
    else
        echo "kafka is running"
    fi
;;
*)
    echo "Usage: $0 {start|stop|restart|status}"
esac
EOF
chmod +x /etc/init.d/kafka
#分别启动 Kafka
service kafka start

#设置开机自启
chkconfig --add kafka

service kafka status
![20240522190542](http://img.shadowwu.club/20240522190542.png)

kafka可视化工具

Kafka-King下载地址 https://github.com/Bronya0/Kafka-King/releases

部署Filebeat+Kafka+ELK

配置filebeat

vim /etc/filebeat/filebeat.yml

filebeat.inputs:
- type: log
#修改为true启用
  enabled: true

  # Paths that should be crawled and fetched. Glob based paths.
  paths:
    #- /var/log/*.log
    - /var/log/messages
    #- c:\programdata\elasticsearch\logs\*
  tags: ["sys"]         #设置索引标签
  fields:           #可以使用 fields 配置选项设置一些参数字段添加到 output 中
    service_name: filebeat
    log_type: syslog
    from: 192.168.126.22
# ------------------------------ Logstash Output -------------------------------
#output.logstash:
 # hosts: ["192.168.126.21:5045"]
output.kafka:
  hosts: ["192.168.126.21:9092", "192.168.126.22:9092", "192.168.126.23:9092"]
  topic: "logs"

systemctl restart filebeat.service

配置logstash

cat > /etc/logstash/pipeconf.d/kafka.conf <<EOF

input {
    kafka {
        bootstrap_servers => "192.168.126.21:9092,192.168.126.22:9092,192.168.126.23:9092,"  #kafka集群地址
        topics  => "logs"     #拉取的kafka的指定topic
        type => "logs_kafka"  #指定 type 字段
        auto_offset_reset => "latest"  #拉取最近数据，earliest为从头开始拉取
        decorate_events => true   #传递给elasticsearch的数据额外增加kafka的属性数据
    }
}

output {
    elasticsearch {
      hosts => ["192.168.126.22:9200"]
      index => "logs_kafka-%{+YYYY.MM.dd}"
  }
}
EOF

增加管道
vim /etc/logstash/pipelines.yml

- pipeline.id: kafka
  path.config: "/etc/logstash/pipeconf.d/kafka.conf"

systemctl restart logstash.service

查看消息

kafka查看生产消费情况


elasticsearch收到数据

问题及解决

Error contacting service. It is probably not running.

[root@k8s-worker02 conf]# service zookeeper status
---------- zookeeper 状态 ------------
/bin/java
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost. Client SSL: false.
Error contacting service. It is probably not running.

原因以下两个文件中有配置出错，主要是server的ip和id不能出错
/usr/local/zookeeper/conf/zoo.cfg
/usr/local/zookeeper/data/myid

kafka节点起不来，无法加入集群

配置文件server.properties不能有注释，id也不能相同
使用service脚本启动不提示错误，但是kafka没有运行，查询还是正在runing状态

kafka获取不到数据

原因是filebeat没有起来，用kafka工具直接生成消息，发现能够被logstash消费，说明是filebeat的问题
必须禁用前一个logstah的output

filebeat只能有一个输出，百度AI误人子弟

#output.logstash:
 # hosts: ["192.168.126.21:5045"]
output.kafka:
  hosts: ["192.168.126.21:9092", "192.168.126.22:9092", "192.168.126.23:9092"]
  topic: "logs"

参考

【ELK 使用指南 3】Zookeeper、Kafka集群与Filebeat+Kafka+ELK架构（附部署实例）
elk+kafka日志分析系统搭建
Kafka基本原理详解（超详细！）