深入浅出系列之 -- Kafka最佳部署实践

阳光穿透心脏的1/2处 2021-10-19 15:38 383阅读 0赞

这是一篇关于 Kafka 实践的文章，内容来自 DataWorks Summit/Hadoop Summit（Hadoop Summit）上的一篇分享，里面讲述了很多关于 Kafka 配置、监控、优化的内容，绝对是在实践中总结出的精华，有很大的借鉴参考意义，本文主要是根据 PPT 的内容进行翻译及适当补充。

Kafka 的架构这里就不多做介绍了，直接步入正题。

# **Kafka 基本配置及性能优化** #

这里主要是 Kafka 集群基本配置的相关内容。

### **硬件要求** ###

Kafka 集群基本硬件的保证

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2N6ejExNDE5Nzk1NzA_size_16_color_FFFFFF_t_70][]

### **OS 调优** ###

>  *  OS page cache：应当可以缓存所有活跃的 **Segment**（Kafka 中最基本的数据存储单位）；
>  *  fd 限制：100k+；
>  *  禁用 swapping：简单来说，swap 作用是**当内存的使用达到一个临界值时就会将内存中的数据移动到 swap 交换空间**，但是此时，内存可能还有很多空余资源，swap 走的是磁盘 IO，对于内存读写很在意的系统，最好禁止使用 swap 分区；
>  *  TCP 调优
>  *  JVM 配置
>     
>      *  JDK 8 并且使用 G1 垃圾收集器
>      *  至少要分配 6-8 GB 的堆内存

### **Kafka 磁盘存储** ###

>  *  使用多块磁盘，并配置为 Kafka 专用的磁盘；
>  *  JBOD vs RAID10；
>  *  JBOD（Just a Bunch of Disks，简单来说它表示一个没有控制软件提供协调控制的磁盘集合，它将多个物理磁盘串联起来，提供一个巨大的逻辑磁盘，数据是按序存储，它的性能与单块磁盘类似）
>  *  JBOD 的一些缺陷：
>     
>      *  任何磁盘的损坏都会导致异常关闭，并且需要较长的时间恢复；
>      *  数据不保证一致性；
>      *  多级目录；
>  *  社区也正在解决这么问题，可以关注 KIP 112、113：
>     
>      *  必要的工具用于管理 JBOD；
>      *  自动化的分区管理；
>      *  磁盘损坏时，Broker 可以将 replicas 迁移到好的磁盘上；
>      *  在同一个 Broker 的磁盘间 reassign replicas；
>  *  RAID 10 的特点：
>     
>      *  可以允许单磁盘的损坏；
>      *  性能和保护；
>      *  不同磁盘间的负载均衡；
>      *  高命中来减少 space；
>      *  单一的 mount point；
>  *  文件系统：
>     
>      *  使用 EXT 或 XFS；
>      *  SSD；

### **基本的监控** ###

> Kafka 集群需要监控的一些指标，这些指标反应了集群的健康度。
> 
>  *  CPU 负载；
>  *  Network Metrics；
>  *  File Handle 使用；
>  *  磁盘空间；
>  *  磁盘 IO 性能；
>  *  GC 信息；
>  *  ZooKeeper 监控。

## **Kafka replica 相关配置及监控** ##

### **Kafka Replication** ###

*  Partition 有两种副本：Leader，Follower；
 *  Leader 负责维护 in-sync-replicas(ISR)
    
     *  `replica.lag.time.max.ms`：默认为10000，如果 follower 落后于 leader 的消息数超过这个数值时，leader 就将 follower 从 isr 列表中移除；
     *  `num.replica.fetchers`，默认为1，用于从 leader 同步数据的 fetcher 线程数；
     *  `min.insync.replica`：Producer 端使用来用于保证 Durability（持久性）；

### **Under Replicated Partitions** ###

当发现 replica 的配置与集群的不同时，一般情况都是集群上的 replica 少于配置数时，可以从以下几个角度来排查问题：

*  JMX 监控项：kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions；
 *  可能的原因：
    
     *  Broker 挂了？
     *  Controller 的问题？
     *  ZooKeeper 的问题？
     *  Network 的问题？
 *  解决办法：
    
     *  调整 ISR 的设置；
     *  Broker 扩容。

### **Controller** ###

*  负责管理 partition 生命周期；
 *  避免 Controller’s ZK 会话超时：
    
     *  ISR 抖动；
     *  ZK Server 性能问题；
     *  Broker 长时间的 GC；
     *  网络 IO 问题；
 *  监控：
    
     *  kafka.controller:type=KafkaController,name=ActiveControllerCount，应该为1；
     *  LeaderElectionRate。

### **Unclean leader 选举** ###

允许不在 isr 中 replica 被选举为 leader。

*  这是 Availability 和 Correctness 之间选择，Kafka 默认选择了可用性；
 *  `unclean.leader.election.enable`：默认为 true，即允许不在 isr 中 replica 选为 leader，这个配置可以全局配置，也可以在 topic 级别配置；
 *  监控：kafka.controller:type=ControllerStats,name=UncleanLeaderElectionsPerSec。

## **Broker 配置** ##

Broker 级别有几个比较重要的配置，一般需要根据实际情况进行相应配置的：

*  `log.retention.{ms, minutes, hours}` , `log.retention.bytes`：数据保存时间；
 *  `message.max.bytes`, `replica.fetch.max.bytes`；
 *  `delete.topic.enable`：默认为 false，是否允许通过 admin tool 来删除 topic；
 *  `unclean.leader.election.enable` = false，参见上面；
 *  `min.insync.replicas` = 2：当 Producer 的 acks 设置为 all 或 -1 时，`min.insync.replicas` 代表了必须进行确认的最小 replica 数，如果不够的话 Producer 将会报 `NotEnoughReplicas` 或 `NotEnoughReplicasAfterAppend` 异常；
 *  `replica.lag.time.max.ms`（超过这个时间没有发送请求的话，follower 将从 isr 中移除）, num.replica.fetchers；
 *  `replica.fetch.response.max.bytes`；
 *  `zookeeper.session.timeout.ms` = 30s；
 *  `num.io.threads`：默认为8，KafkaRequestHandlerPool 的大小。

## **Kafka 相关资源的评估** ##

### **集群评估** ###

*  Broker 评估
    
     *  每个 Broker 的 Partition 数不应该超过2k；
     *  控制 partition 大小（不要超过25GB）；
 *  集群评估（Broker 的数量根据以下条件配置）
    
     *  数据保留时间；
     *  集群的流量大小；
 *  集群扩容：
    
     *  磁盘使用率应该在 60% 以下；
     *  网络使用率应该在 75% 以下；
 *  集群监控
    
     *  保持负载均衡；
     *  确保 topic 的 partition 均匀分布在所有 Broker 上；
     *  确保集群的阶段没有耗尽磁盘或带宽。

### **Broker 监控** ###

*  Partition 数：kafka.server:type=ReplicaManager,name=PartitionCount；
 *  Leader 副本数：kafka.server:type=ReplicaManager,name=LeaderCount；
 *  ISR 扩容/缩容率：kafka.server:type=ReplicaManager,name=IsrExpandsPerSec；
 *  读写速率：Message in rate/Byte in rate/Byte out rate；
 *  网络请求的平均空闲率：NetworkProcessorAvgIdlePercent；
 *  请求处理平均空闲率：RequestHandlerAvgIdlePercent。

### **Topic 评估** ###

*  partition 数
    
     *  Partition 数应该至少与最大 consumer group 中 consumer 线程数一致；
     *  对于使用频繁的 topic，应该设置更多的 partition；
     *  控制 partition 的大小（25GB 左右）；
     *  考虑应用未来的增长（可以使用一种机制进行自动扩容）；
 *  使用带 key 的 topic；
 *  partition 扩容：当 partition 的数据量超过一个阈值时应该自动扩容（实际上还应该考虑网络流量）。

### **合理地设置 partition** ###

*  根据吞吐量的要求设置 partition 数：
    
     *  假设 Producer 单 partition 的吞吐量为 P；
     *  consumer 消费一个 partition 的吞吐量为 C；
     *  而要求的吞吐量为 T；
     *  那么 partition 数至少应该大于 T/P、T/C 的最大值；
 *  更多的 partition，意味着：
    
     *  更多的 fd；
     *  可能增加 Unavailability（可能会增加不可用的时间）；
     *  可能增加端到端的延迟；
     *  client 端将会使用更多的内存。

**Partition 的增加将会带来以下几个优点和缺点：**

*  增加吞吐量：对于 consumer 来说，一个 partition 只能被一个 consumer 线程所消费，适当增加 partition 数，可以增加 consumer 的并发，进而增加系统的吞吐量；
 *  需要更多的 fd：对于每一个 segment，在 broker 都会有一个对应的 index 和实际数据文件，而对于 Kafka Broker，它将会对于每个 segment 每个 index 和数据文件都会打开相应的 file handle（可以理解为 fd），因此，partition 越多，将会带来更多的 fd；
 *  可能会增加数据不可用性（主要是指增加不可用时间）：主要是指 broker 宕机的情况，越多的 partition 将会意味着越多的 partition 需要 leader 选举（leader 在宕机这台 broker 的 partition 需要重新选举），特别是如果刚好 controller 宕机，重新选举的 controller 将会首先读取所有 partition 的 metadata，然后才进行相应的 leader 选举，这将会带来更大不可用时间；
 *  可能增加 End-to-end 延迟：一条消息只有其被同步到 isr 的所有 broker 上后，才能被消费，partition 越多，不同节点之间同步就越多，这可能会带来毫秒级甚至数十毫秒级的延迟；
 *  Client 将会需要更多的内存：Producer 和 Consumer 都会按照 partition 去缓存数据，每个 partition 都会带来数十 KB 的消耗，partition 越多, Client 将会占用更多的内存。

## **Producer 的相关配置、性能调优及监控** ##

### **Quotas** ###

*  避免被恶意 Client 攻击，保证 SLA；
 *  设置 produce 和 fetch 请求的字节速率阈值；
 *  可以应用在 user、client-id、或者 user 和 client-id groups；
 *  Broker 端的 metrics 监控：throttle-rate、byte-rate；
 *  `replica.fetch.response.max.bytes`：用于限制 replica 拉取请求的内存使用；
 *  进行数据迁移时限制贷款的使用，`kafka-reassign-partitions.sh -- -throttle option`。

### **Kafka Producer** ###

*  使用 Java 版的 Client；
 *  使用 `kafka-producer-perf-test.sh` 测试你的环境；
 *  设置内存、CPU、batch 压缩；
    
     *  batch.size：该值设置越大，吞吐越大，但延迟也会越大；
     *  linger.ms：表示 batch 的超时时间，该值越大，吞吐越大、但延迟也会越大；
     *  `max.in.flight.requests.per.connection`：默认为5，表示 client 在 blocking 之前向单个连接（broker）发送的未确认请求的最大数，超过1时，将会影响数据的顺序性；
     *  `compression.type`：压缩设置，会提高吞吐量；
     *  `acks`：数据 durability 的设置；
 *  避免大消息
    
     *  会使用更多的内存；
     *  降低 Broker 的处理速度；

### **性能调优** ###

*  如果吞吐量小于网络带宽
    
     *  增加线程；
     *  提高 batch.size；
     *  增加更多 producer 实例；
     *  增加 partition 数；
 *  设置 acks=-1 时，如果延迟增大：可以增大 `num.replica.fetchers`（follower 同步数据的线程数）来调解；
 *  跨数据中心的传输：增加 socket 缓冲区设置以及 OS tcp 缓冲区设置。

### **Prodcuer 监控** ###

*  batch-size-avg
 *  compression-rate-avg
 *  waiting-threads
 *  buffer-available-bytes
 *  record-queue-time-max
 *  record-send-rate
 *  records-per-request-avg

## **Kafka Consumer 配置、性能调优及监控** ##

### **Kafka Consumer** ###

*  使用 `kafka-consumer-perf-test.sh` 测试环境；
 *  吞吐量问题：
    
     *  partition 数太少；
     *  OS page cache：分配足够的内存来缓存数据；
     *  应用的处理逻辑；
 *  offset topic（`__consumer_offsets`）
    
     *  `offsets.topic.replication.factor`：默认为3；
     *  `offsets.retention.minutes`：默认为1440，即 1day；  
        – MonitorISR，topicsize；
 *  offset commit较慢：异步 commit 或 手动 commit。

### **Consumer 配置** ###

*  `fetch.min.bytes` 、`fetch.max.wait.ms`；
 *  `max.poll.interval.ms`：调用 `poll()` 之后延迟的最大时间，超过这个时间没有调用 `poll()`的话，就会认为这个 consumer 挂掉了，将会进行 rebalance；
 *  `max.poll.records`：当调用 `poll()` 之后返回最大的 record 数，默认为500；
 *  `session.timeout.ms`；
 *  Consumer Rebalance  
    – check timeouts  
    – check processing times/logic  
    – GC Issues
 *  网络配置；

### **Consumer 监控** ###

### Consumer 是否跟得上数据的发送速度。 ###

*  Consumer Lag：consumer offset 与 the end of log（partition 可以消费的最大 offset） 的差值；
 *  监控
    
     *  metric 监控：records-lag-max；
     *  通过 `bin/kafka-consumer-groups.sh` 查看；
     *  用于 consumer 监控的 LinkedIn’s Burrow；
 *  减少 Lag
    
     *  分析 consumer：是 GC 问题还是 Consumer hang 住了；
     *  增加 Consumer 的线程；
     *  增加分区数和 consumer 线程；

## **如何保证数据不丢** ##

这个是常用的配置：

*  Producer
    
     *  block.on.buffer.full=true
     *  retries=Long.MAX\_VALUE
     *  acks=all
     *  max.in.flinght.requests.per.connection=1
     *  close producer
 *  Broker
    
     *  relication factor >= 3
     *  min.insync.replicas=2
     *  disable unclean leader election

*  Consumer
    
     *  min.insync.replicas=2
     *  disable auto.buffer.full=true
     *  disable auto.buffer.full=true
     *  commit offsets only after the message are processed

*  `block.on.buffer.full`：默认设置为 false，当达到内存设置时，可能通过 block 停止接受新的 record 或者抛出一些错误，默认情况下，Producer 将不会抛出  BufferExhaustException，而是当达到 `max.block.ms` 这个时间后直接抛出 TimeoutException。设置为 true 的意义就是将 `max.block.ms` 设置为 Long.MAX\_VALUE，未来版本中这个设置将被遗弃，推荐设置 `max.block.ms`。

\-----------------------------------------------------------------------------------------------------------------------------------------------------------------------

考虑一千次，不如去做一次；犹豫一万次，不如实践一次；华丽的跌倒，胜过无谓的彷徨，将来的你，一定会感谢现在奋斗的你。欢迎大家加入大数据交流群：725967421     一起交流，一起进步！！

\------------------------------------------------------------------------------------------------------------------------------------------------------------------------

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2N6ejExNDE5Nzk1NzA_size_16_color_FFFFFF_t_70]: /images/20211019/fa2b3694e42d4c1ebd7394f0d85f214d.png