文章目录
- kafka+ELK连接
- 部署Kafka
- kafka操作命令
- kafka架构深入
- Filebeat+Kafka+ELK连接
kafka+ELK连接
部署Kafka
###关闭防火墙
systemctl stop firewalld
systemctl disable firewalld
setenforce 0
vim /etc/selinux/config
SELINUX=disabled
###下载安装包
官方下载地址：http://kafka.apache.org/downloads.html
cd /opt
wget https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.7.1/kafka_2.13-2.7.1.tgz
###安装 Kafka
cd /opt/
tar xf kafka_2.13-2.7.1.tgz
mv kafka_2.13-2.7.1 /usr/local/kafka
###修改配置文件
##备份配置文件
cd /usr/local/kafka/config/
cp server.properties{,.bak}
vim server.properties
---21行--
broker.id=0
###21行，broker的全局唯一编号，每个broker不能重复，因此要在其他机器上配置 broker.id=1、broker.id=2
---31行---
listeners=PLAINTEXT://192.168.242.70:9092
###31行，指定监听的IP和端口，如果修改每个broker的IP需区分开来，也可保持默认配置不用修改
---42行---
num.network.threads=3
###42行，broker 处理网络请求的线程数量，一般情况下不需要去修改
---45行---
num.io.threads=8
#45行，用来处理磁盘IO的线程数量，数值应该大于硬盘数
---48行---
socket.send.buffer.bytes=102400
#48行，发送套接字的缓冲区大小
---51行---
socket.receive.buffer.bytes=102400
#51行，接收套接字的缓冲区大小
---54行---
socket.request.max.bytes=104857600
#54行，请求套接字的缓冲区大小
---60行---
log.dirs=/usr/local/kafka/logs
#60行，kafka运行日志存放的路径，也是数据存放的路径
---65行---
num.partitions=1
#65行，topic在当前broker上的默认分区个数，会被topic创建时的指定参数覆盖
---69行---
num.recovery.threads.per.data.dir=1
#69行，用来恢复和清理data下数据的线程数量
---103行---
log.retention.hours=168
#103行，segment文件（数据文件）保留的最长时间，单位为小时，默认为7天，超时将被删除
---110行---
log.segment.bytes=1073741824
#110行，一个segment文件最大的大小，默认为 1G，超出将新建一个新的segment文件
---123行---
zookeeper.connect=192.168.242.70:2181,192.168.242.71:2181,192.168.242.72:2181
###123行，配置连接Zookeeper集群地址
###修改环境变量
vim /etc/profile
export KAFKA_HOME=/usr/local/kafka
export PATH=$PATH:$KAFKA_HOME/bin
###加载配置
source /etc/profile
###复制文件到zookeeper服务器哦
scp -r /usr/local/kafka 192.168.242.71:/usr/local
scp -r /usr/local/kafka 192.168.242.72:/usr/local
###部署kafka启动脚本
vim /etc/init.d/kafka
#!/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
###设置开机自启
cd /etc/init.d/
chmod +x /etc/init.d/kafka
chkconfig --add kafka
###分别启动 Kafka
service kafka start
###另外一种启动方式
cd /usr/local/kafka/bin
./kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties
netstat -lntp | grep 9092
kafka操作命令
####创建topic
kafka-topics.sh --create --zookeeper 192.168.242.70:2181,192.168.242.71:2181,192.168.242.72:2181 --replication-factor 2 --partitions 3 --topic test
--zookeeper：定义 zookeeper 集群服务器地址，如果有多个 IP 地址使用逗号分割，一般使用一个 IP 即可
--replication-factor：定义分区副本数，1 代表单副本，建议为 2
--partitions：定义分区数
--topic：定义 topic 名称
###查看当前服务器中的所有 topic
kafka-topics.sh --list --zookeeper 192.168.242.70:2181,192.168.242.71:2181,192.168.242.72:2181
###查看某个 topic 的详情
kafka-topics.sh --describe --zookeeper 192.168.242.70:2181,192.168.242.71:2181,192.168.242.72:2181
###发布消息
kafka-console-producer.sh --broker-list 192.168.242.70:9092,192.168.242.71:9092,192.168.242.72:9092 --topic test
###消费消息
kafka-console-consumer.sh --bootstrap-server 192.168.242.70:9092,192.168.242.71:9092,192.168.242.72:9092 --topic test --from-beginning
--from-beginning：会把主题中以往所有的数据都读取出来
###修改分区数
kafka-topics.sh --zookeeper 192.168.242.70:2181,192.168.242.71:2181,192.168.242.72:2181 --alter --topic test --partitions 6
###删除 topic
kafka-topics.sh --delete --zookeeper 192.168.242.70:2181,192.168.242.71:2181,192.168.242.72:2181 --topic test
kafka架构深入
-
Kafka 工作流程及文件存储机制
- Kafka 中消息是以 topic 进行分类的，生产者生产消息，消费者消费消息，都是面向 topic 的。
-
topic 是逻辑上的概念，而 partition 是物理上的概念，每个 partition 对应于一个 log 文件，该 log 文件中存储的就是 producer 生产的数据。
-
Producer 生产的数据会被不断追加到该 log 文件末端，且每条数据都有自己的 offset。
-
消费者组中的每个消费者，都会实时记录自己消费到了哪个 offset，以便出错恢复时，从上次的位置继续消费。
-
由于生产者生产的消息会不断追加到 log 文件末尾，为防止 log 文件过大导致数据定位效率低下，Kafka 采取了分片和索引机制，将每个 partition 分为多个 segment。
-
每个 segment 对应两个文件：“.index” 文件和 “.log” 文件。
-
这些文件位于一个文件夹下，该文件夹的命名规则为：topic名称+分区序号。例如，test 这个 topic 有三个分区， 则其对应的文件夹为 test-0、test-1、test-2。
-
index 和 log 文件以当前 segment 的第一条消息的 offset 命名。
- “.index” 文件存储大量的索引信息，
- “.log” 文件存储大量的数据，
- 索引文件中的元数据指向对应数据文件中 message 的物理偏移地址。
-
数据可靠性保证
- 为保证 producer 发送的数据，能可靠的发送到指定的 topic，topic 的每个 partition 收到 producer 发送的数据后， 都需要向 producer 发送 ack（acknowledgement 确认收到），如果 producer 收到 ack，就会进行下一轮的发送，否则重新发送数据。
-
数据一致性问题
- LEO：指的是每个副本最大的 offset；
- HW：指的是消费者能见到的最大的 offset，所有副本中最小的 LEO。
-
follower 故障
- follower 发生故障后会被临时踢出 ISR（Leader 维护的一个和 Leader 保持同步的 Follower 集合），待该 follower 恢复后，follower 会读取本地磁盘记录的上次的 HW，并将 log 文件高于 HW 的部分截取掉，从 HW 开始向 leader 进行同步。等该 follower 的 LEO 大于等于该 Partition 的 HW，即 follower 追上 leader 之后，就可以重新加入 ISR 了。
-
leader 故障
- leader 发生故障之后，会从 ISR 中选出一个新的 leader， 之后，为保证多个副本之间的数据一致性，其余的 follower 会先将各自的 log 文件高于 HW 的部分截掉，然后从新的 leader 同步数据。
-
注：这只能保证副本之间的数据一致性，并不能保证数据不丢失或者不重复。
-
ack 应答机制
- 对于某些不太重要的数据，对数据的可靠性要求不是很高，能够容忍数据的少量丢失，所以没必要等 ISR 中的 follower 全部接收成功。
- 所以 Kafka 为用户提供了三种可靠性级别，用户根据对可靠性和延迟的要求进行权衡选择。
-
当 producer 向 leader 发送数据时，可以通过 request.required.acks 参数来设置数据可靠性的级别：
- 0：这意味着producer无需等待来自broker的确认而继续发送下一批消息。这种情况下数据传输效率最高，但是数据可靠性确是最低的。当broker故障时有可能丢失数据。
- 1（默认配置）：这意味着producer在ISR中的leader已成功收到的数据并得到确认后发送下一条message。如果在follower同步成功之前leader故障，那么将会丢失数据。
- -1（或者是all）：producer需要等待ISR中的所有follower都确认接收到数据后才算一次发送完成，可靠性最高。但是如果在 follower 同步完成后，broker 发送ack 之前，leader 发生故障，那么会造成数据重复。
-
三种机制性能依次递减，数据可靠性依次递增。
注：在 0.11 版本以前的Kafka，对此是无能为力的，只能保证数据不丢失，再在下游消费者对数据做全局去重。在 0.11 及以后版本的 Kafka，引入了一项重大特性：幂等性。所谓的幂等性就是指 Producer 不论向 Server 发送多少次重复数据， Server 端都只会持久化一条。
+ kafka会通过ack机制保证数据的可靠性
- ack配置参数有
- 0（效果类似异步复制）：不等待follower同步完成就让生产者发送下一条消息
- 1（效果类似半同步复制）：至少等待一个follower同步完成才让生产者发送下一条消息
- -1（效果类似全同步复制）：等待所有follower同步完成才让生产者发送下一条消息
Filebeat+Kafka+ELK连接
###部署 Zookeeper+Kafka 集群
###部署 Filebeat，修改配置文件
cd /usr/local/filebeat
vim filebeat.yml
filebeat.prospectors:
- type: log
enabled: true
paths:
- /var/log/httpd/access_log
tags: ["access"]
- type: log
enabled: true
paths:
- /var/log/httpd/error_log
tags: ["error"]
......
#添加输出到 Kafka 的配置
output.kafka:
enabled: true
hosts: ["192.168.242.70:9092","192.168.242.71:9092","192.168.242.72:9092"] #指定 Kafka 集群配置
topic: "httpd" #指定 Kafka 的 topic
###启动 filebeat
./filebeat -e -c filebeat.yml
###部署 ELK，在 Logstash 组件所在节点上新建一个 Logstash 配置文件
cd /etc/logstash/conf.d/
vim kafka.conf
input {
kafka {
bootstrap_servers => "192.168.242.70:9092,192.168.242.71:9092,192.168.242.72:9092"
#kafka集群地址
topics => "httpd" #拉取的kafka的指定topic
type => "httpd_kafka" #指定 type 字段
codec => "json" #解析json格式的日志数据
auto_offset_reset => "latest" #拉取最近数据，earliest为从头开始拉取
decorate_events => true #传递给elasticsearch的数据额外增加kafka的属性数据
}
}
output {
if "access" in [tags] {
elasticsearch {
hosts => ["192.168.242.66:9200"]
index => "httpd_access-%{+YYYY.MM.dd}"
}
}
if "error" in [tags] {
elasticsearch {
hosts => ["192.168.242.66:9200"]
index => "httpd_error-%{+YYYY.MM.dd}"
}
}
stdout { codec => rubydebug }
}
####启动 logstash
logstash -f kafka.conf
