Kafka Broker

Kafka 副本

副本 基本信息

1. Kafka副本作用：提高数据可靠性
2. Kafka默认副本1个，生产环境一般配置为2个,保证数据可靠性；太多副本会增加磁盘存储空间，增加网络上数据传输，降低效率
3. Kafka中副本分为：Leader和Follower。Kafka生产者只会把数据发往Leader，然后Follower找Leader进行同步数据
4. Kafka分区中的所有副本统称为AR(Assigned Repllicas)

• AR=ISR+OSR
• ISR:表示和Leader保持同步的Follower集合。如果Follower长时间未向Leader发送通信请求或同步数据，则该Follower将被踢出ISR。该时间阀值由replica.lag.time.max.ms参数设定，默认30s。Leader发生故障之后，就会从ISR中选举新的Leader
• OSR,表示Follower与Leader副本同步时，延迟过多的副本

Zookeeper存储的Kafka信息

1. 启动Zookeeper客户端

   bin/zkCli.sh
   

2. 通过ls命令查看kafka相关信息

   ls /kafka
   

3. Zookeeper中存储的Kafka信息

   1. /kafka/brokers/ids:[0,1,2]：记录有哪些服务器
   2. /kafka/borkers/topics/first/partitions/0/state:{"leader":1,"isr":[1,0,2]}：记录谁是Leader，有哪些服务器可用
   3. /kafka/controller：{"brokerid":0}：辅助选举Leader

Kafka Broker总体工作流程

Leader选举流程

• AR(Assigned Replica):Kafka分区中的所有副本统称
  • AR 是指 Kafka 集群中一个分区所对应的所有副本的集合。
  • 一个分区可以有多个副本，其中包括一个 Leader 副本和若干个 Follower 副本。
  • AR 包含了 Leader 副本和 Follower 副本。
• ISR(In-Sync Replica):同步副本集
  • ISR 是指分区的 Leader 副本和正在与 Leader 副本保持同步的 Follower 副本的集合。
  • Follower 副本只有在与 Leader 副本保持同步时，才被认为是 ISR 的一部分。
  • ISR 中的副本会参与消息的复制和读取操作。

1. broker启动后在zk（/brokers/ids）中注册
2. controller谁先注册，谁说了算
3. 由选举出来的Controller监听brokers节点变化
4. Controller决定Leader选举
   • 选举规则：在isr中存活为前提，按照AR中排在前面的优先。例如：ar[1,0,2],isr[1,0,2],那么leader会按照1,0,2的顺序轮询
5. Controller将节点信息上传到ZK(/kafka/borkers/topics/first/partitions/0/state)
6. 其他contorller从zk同步相关信息
7. 假设Broker1中Leader挂了
8. Controller监听到节点变化
9. 获取ISR
10. 选举新的Leader（在isr中存活为前提，按照AR中排在前面的优先）
11. 更新Leader及ISR

Leader和Follower故障处理细节

Follower故障处理细节

• LEO(Log End Offset)：每个副本的最后一个offset,LEO是最新的offset+1
• HW(High Watermark)：所有副本中最小的LEO
  1. Follower故障
     1. Follower发生故障后会被临时踢出ISR
     2. 这个期间Leader和Follower继续接收数据
     3. 待该Follower恢复后，Follower会读取本地磁盘记录的上次的HW，并将log文件高于HW的部分截取掉，从HW开始向Leader进行同步
     4. 等该Follower的LEO大于等于该Partition的HW，即Follower追上Leader之后，就可以重新加入ISR了

Leader故障处理细节

1. Leader故障

   1. Leader发生故障之后，会从ISR中选出一个新的Leader
   2. 为保证多个副本之间的数据一致性，其余的Follower会先将各自的log文件高于HW的部分截掉，然后从新的Leader同步数据
   • 注意：这只能保证副本之间的数据一致性，并不能保证数据不丢失或者不重复

文件存储

文件存储机制

Topic数据的存储机制

• Topic是逻辑上的概念，而partition是物理上的概念，每个partition对应于一个log文件，该log文件中存储的就是Producer生产的数据。Producer生产的数据会被不断追加到该log文件末端，为防止log文件过大导致数据定位效率底下，Kafka采取了分片和索引机制，将每个partition分为多个segment。每个segment包括:".index文件"、".log"文件和".timeindex"等文件。这些文件位于一个文件夹下，该文件夹的命名规则为：topic名称+分区序号，例如：first-0

Topic数据存储位置

1. 查看hadoop102的/opt/module/kafka/datas/first-1路径上的文件

   cd /opt/module/kafka/datas/first-0
   ls
   

2. 直接查看log日志，发现是乱码

   cat 00000000000000000014.log
   

3. 通过工具查看index和log信息

   kafka-run-class.sh kafka.tools.DumpLogSegments --files ./00000000000000000014.index
   
   offset: 115 position: 4165
   offset: 202 position: 8296
   offset: 292 position: 12461
   offset: 374 position: 16614
   
   kafka-run-class.sh kafka.tools.DumpLogSegments --files ./00000000000000000014.log
   

index文件和log文件详解

• 注意
  1. index为稀疏索引，大约每往log文件写入4kb数据，会往index文件写入一条索引。参数log.index.interval.bytes默认4kb
  2. index文件中保存的offset为相对offset，这样能确保offset的值所占空间不会过大，因此能将offset的值控制在固定大小
• 步骤
  1. 根据目标offset定位Segment文件
  2. 找到小于等于目标offset的最大offset对应的索引项
  3. 定位到log文件
  4. 向下遍历找到目标Record

日志存储参数配置

参数	描述
log.segment.bytes	Kafka中log日志是分成一块块存储的，此配置是指log日志划分成块的大小，默认值1G
log.index.interval.bytes	默认4kb，kafka里面每当写入了4kb大小的日志(.log)，然后就往index文件里面记录一条索引。稀疏索引

文件清理策略

• Kafka中默认的日志保存时间为7天，可以通过调整如下参数修改保存时间
  • log.retention.hours：最低优先级小时，默认7天
  • log.retention.minutes：分钟
  • log.retention.ms：最高优先级毫秒
  • log.retention.check.interval.ms：负责设置检查周期，默认5分钟
• Kafka中提供的日志清理策略有delete和compact两种
  1. delete日志删除：将过期数据删除
     • log.cleanup.policy=delete：所有数据启用删除策略
       1. 基于时间：默认打开。以segment中所有记录中的最大时间戳作为该文件时间戳
       2. 基于大小：默认关闭。超过设置的所有日志总大小，删除最早的segment。
          • log.retention.bytes：默认等于-1，表示无穷大
  2. compact日志压缩
     • compact日志压缩：对于相同key的不同value值，只保留最后一个版本
       • log.cleanup.policy=compact：所有数据启用压缩策略
     • 压缩后的offset可能是不连续的，比如上图中没有6，当从这些offset消费消息时，将会拿到比这个offset大的offset对应的消息，实际上会拿到offset为7的消息，并从这个位置开始消费
     • 这种策略只适合特殊场景，比如消息的key是用户ID,value是用户的资料，通过这种压缩策略，整个消息集里就保存了所有用户最新的资料

高效读写数据

1. Kafka本身是分布式集群，可以采用分区技术，并行度高
2. 读数据采用稀疏索引，可以快速定位要消费的数据
3. 顺序写磁盘
   • Kafka的producer生产数据，要写入到log文件中，写的过程是一直追加到文件末端，为顺序写。
4. 页缓存+零拷贝技术
   • 零拷贝：Kafka的数据加工处理操作交由Kafka生产者和Kafka消费者处理。Kafka Broker应用层不关心存储的数据，所以就不用走应用层，传输效率高
   • PageCache页缓存：Kafka重度依赖底层操作系统提供的PageCache功能。当上层有写操作时，操作系统只是将数据写入PageCache。当读操作发生时，先从PageCache中查找，如果找不到，再去磁盘中读取。实际上PageCache是把尽可能多的空闲内存都当做了磁盘缓存来使用

参数

参数	描述
log.flush.interval.messages	强制页缓存刷写到磁盘的的条数，默认是long的最大值，9223372036854775807。一般不建议修改，交给系统自己管理
log.flush.interval.ms	每隔多久，刷数据到磁盘，默认是null。一般不建议修改，交给系统自己管理。

Broker重要参数

参数名称	描述
replica.lag.time.max.ms	ISR中 如果 Follower长时间未向 Leader 发送通 信请求或同步数据，则该 Follower 将被踢出 ISR 。 该时间阈值 默认 30s 。
auto.leader.rebalance.enable	默认是 true 。 自动 Leader Partition 平衡 。
leader.imbalance.per.broker.percentage	默认是 1 0% 。 每个 broker 允许的不平衡的 leader 的比率。如果每个 broker 超过了这个值，控制器 会触发 leader 的平衡。
leader.imbalance.check.interval.seconds	默认值 300 秒 。检查 leader 负载是否平衡的间隔时间。
log.segment.bytes	Kafka 中 log 日志是分成一块块存储的，此配置是 指 log 日志划分 成块的大小 默认值 1G 。
log.index.interval.bytes	默认 4kb ,kafka 里面每当写入了 4kb 大小的日志（.log ）， 然后就往 index文件里面记录一个索引 。
log.retention.hours	Kafka中数据保存的时间， 默认 7 天。
log.retention.minutes	Kafka 中数据保存的时间， 分钟级别 ，默认关闭。
log.retention.ms	Kafka 中数据保存的时间， 毫秒级别 ，默认关闭。
log.retention.check.interval.ms	检查数据是否保存超时的间隔，默认是 5 分钟
log.retention.bytes	默认等于 1 ，表示无穷大。 超过设置的 所有 日志总大小，删除最早的 segment。
log.cleanup.policy	默认是 delete ，表示所有数据启用删除策略 如果设置值为 compact ，表示所有数据启用压缩策略。
num.io.threads	默认是8 。 负责写磁盘的线程数。整个参数值要占总核数的 5 0% 。
num .replica.fetchers	副本拉取线程数，这个参数占总核数的 50% 的 1/3
num.network.threads	默认是3 。 数据传输线程数，这个参数占总核数的 5 0% 的 2 /3 。
log.flush<br/>.interval.messages	强制页缓存刷写到磁盘的条数，默认是 long 的最大值， 9223372036854775807 。一般不建议修改，交给系统自己管理。
log.flush.interval.ms	每隔多久，刷数据到磁盘，默认是 null 。一般不建议修改，交给系统自己管理。

生产经验

节点服役和退役

服役新节点

新节点准备

1. 关闭hadoop104,并右键执行克隆操作

2. 开启hadoop105,并修改IP地址

   vim   /etc/sysconfig/network-scripts/ifcfg-ens33
   
   DEVICE=ens33
   TYPE=Ethernet 
   ONBOOT=yes 
   BOOTPROTO=static 
   NAME="ens33" 
   IPADDR=192.168.10.105 
   PREFIX=24 
   GATEWAY=192.168.10.2
   DNS1=192.168.10.2
   

3. 在 hadoop 105 上，修改主机名称为 hadoop 105

    vim /etc/hostname
    hadoop105
   

4. 重新启动 hadoop 104 、 hadoop 105。

5. 修改 hadoop105中 kafka 的 broker .id 为 3 。

6. 删除 hadoop 105 中 kafka 下的 datas 和 logs。

   rm rf datas/* logs/*
   

7. 启动 hadoop 102 、 hadoop 103 、 hadoop 104 上的 kafka 集群 。

   zk.sh start
   kf.sh start
   

8. 单独启动 hadoop 105 中的 kafka

   bin/kafka-server-start.sh -daemon ./config/server.properties
   

执行 负载均衡 操作

1. 创建一个要均衡的主题

   vim topics-to-move.json
   {
   	"topics": [
   
   		{"topic": "first"}
   	],
   	"version": 1
   }
   

2. 生成一个负载均衡的计划 。

   bin/kafka-reassign-partitions.sh  --bootstrap-server hadoop102:9092 --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2,3" --generate
   
   Current partition replica assignment
   {"version":1,"partitions":[{"topic":"first","partition":0,"replic as":[0,2,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[2,1,0],"log_dirs":["any","any","any"]},{"to
   pic":"first","partition":2,"replicas":[1,0,2],"log_dirs":["any"," any","any"]}]}
   
   Proposed partition reassignment configuration
   {"version":1,"partitions":[{"topic":"first","partition":0,"replic as":[2,3,0],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[3,0,1],"log_dirs":["any","any","any"]},{"to
   pic":"first","partition":2,"replicas":[0,1,2],"log_dirs":["any","any","any"]}]}
   
   

3. 创建副本存储计划（所有副本存储在 broker 0 、 broker 1 、 broker 2 、 broker 3 中） 。

   vim increase-replication-factor.json
   
   #输入如下内容
   {"version":1,"partitions":[{"topic":"first","partition":0,"replic as":[2,3,0],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[3,0,1],"log_dirs":["any","any","any"]},{"to
   pic":"first","partition":2,"replicas":[0,1,2],"log_dirs":["any","any","any"]}]}
   

4. 执行副本存储计划

   bin/kafka-reassign-partitions.sh  --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute
   

5. 验证副本存储计划

   bin/kafka-reassign-partitions.sh  --bootstrap-server hadoop102:9092 --reassignment-json-file
   increase-replication-factor.json --verify
   

退役旧节点

执行负载均衡操作

• 先按照退役一台节点，生成执行计划，然后按照服役时操作流程执行负载均衡。

1. 创建一个要均衡的主题

   vim topics-to-move.json
   
   {
   	"topics": [
   
   		{"topic": "first"}
   	],
   	"version": 1
   }
   
   

2. 创建执行计划

   bin/kafka-reassign-partitions.sh  --bootstrap-server hadoop102:9092 --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2" --generate
   
   Current partition replica assignment
   {"version":1,"partitions":[{"topic":"first","partition":0,"replic as":[2,0,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[3,1,2],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[0,2,3],"log_dirs":["any","any","any"]}]}
   
   Proposed partition reassignment configuration
   {"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2,0,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[1,2,0],"log_dirs":["any","any","any"]}]}
   
   

3. 创建副本存储计划（所有副本存储在 broker0、broker1、broker2 中）。

   vim increase-replication-factor.json
   
   {"version":1,"partitions":[{"topic":"first","partition":0,"replic as":[2,0,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[1,2,0],"log_dirs":["any","any","any"]}]}
   
   

4. 执行副本存储计划。

   bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute
   

5. 验证副本存储计划。

   bin/kafka-reassign-partitions.sh  --bootstrap-server hadoop102:9092 --reassignment-json-file increase
   -replication-factor.json --verify
   
   Status of partition reassignment:
   Reassignment of partition first-0 Reassignment of partition first-1 Reassignment of partition 
   first-2
   is complete. is complete. is complete.
   
   Clearing broker-level throttles on brokers 0,1,2,3
   Clearing topic-level throttles on topic first
   

执行停止命令

• 在 hadoop105上执行停止命令即可 。

  bin/kafka-server-stop.sh
  

手动调整分区副本存储

• 在生产环境中，每台服务器的配置和性能不一致，但是Kafka只会根据自己的代码规则创建对应的分区副本，就会导致个别服务器存储压力较大。

• 手动调整分区副本存储的步骤

  1. 创建一个新的topic，名称为three

     bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --topic three --partitions 4 --replication-factor 2
     

  2. 查看分区副本存储情况

     bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic three
     

  3. 创建副本存储计划

     vim increase-replication-factor.json
     
     {
     "version":1, "partitions":[{"topic":"three","partition":0,"replicas":[0,1]},
     {"topic":"three","partition":1,"replicas":[0,1]},
     {"topic":"three","partition":2,"replicas":[1,0]},
     {"topic":"three","partition":3,"replicas":[1,0]}]
     }
     
     

  4. 执行副本存储计划

     bin/kafka-reassign-partitions.sh  --bootstrap-server hadoop102:9092  --reassignment-json-file increase-replication-factor.json --execute
     

  5. 验证副本存储计划

     bin/kafka-reassign-partitions.sh  --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --verify
     

  6. 查看分区副本存储情况

     bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic three
     

Leader Partition负载平衡

• 正常情况下，Kafka本身会自动把Leader Partition均匀分散在各个机器上，来保证每台机器的读写吞吐量都是均匀的。但是如果某些broker宕机，会导致Leader Partition过于集中在其他少部分几台broker上，这会导致少数几台broker的读写请求压力过高，其他宕机的broker重启之后都是follow partition，读写请求很低，造成集群负载不均匀

Leader Partition负载平衡参数配置

参数名称	描述
auto.leader.rebalance.enable	默认是true。自动Leader Partition平衡。生产环境中，leader重选举的代价比较大，可能会带来性能影响，建议设置为false关闭
leader.imbalance.per.broker.percentage	默认是10%。每个broker允许的不平衡的leader的比率。如果每个broker超过了这个值，控制器会触发leader的平衡
leader.imbalance.check.interval.seconds	默认值300秒。检查leader负载是否平衡的间隔时间

增加副本因子

• 在生产环境当中，由于某个主题的重要等级需要提升，需要考虑增加副本。副本数的增加需要先制定计划，然后根据计划执行

  1. 创建topic

     bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --topic four --partitions 3 --replication-factor 1
     

  2. 手动增加副本存储

     1. 创建副本存储计划

        vim increase-replication-factor.json
        
        {"version":1,"partitions":[{"topic":"four","partition":0,"replicas":[0,1,2]},{"topic":"four","partition":1,"replicas":[0,1,2]},{"topic":"four","partition":2,"replicas":[0,1,2]}]}}
        
        

     2. 执行副本存储计划

        bin/kafka-reassign-partitions.sh  --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute