[譯] Kafka 最佳實踐

ApacheKafka · 發表 2019-01-30 08:47:26

摘要：這裡翻譯一篇關於 Kafka 實踐的文章，內容來自 DataWorks Summit/Hadoop Summit 上一篇分享，PPT 參考 https://www.slideshare.net/HadoopSummit/apache-kafka-best-practices 裡面講述了很...

這裡翻譯一篇關於 Kafka 實踐的文章，內容來自 DataWorks Summit/Hadoop Summit 上一篇分享，PPT 參考 https://www.slideshare.net/HadoopSummit/apache-kafka-best-practices 裡面講述了很多關於 Kafka 配置、監控、優化的內容，絕對是在實踐中總結出的精華，有很大的借鑑參考意義，本文主要是根據 PPT 的內容進行翻譯及適當補充。

Kafka 的架構這裡就不多做介紹了，直接不如正題。

Kafka 基本配置及效能優化

這裡主要是 Kafka 叢集基本配置的相關內容。

硬體要求

Kafka 叢集基本硬體的保證

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 的設定可以參考這篇文章 https://www.confluent.io/blog/how-choose-number-topics-partitions-kafka-cluster ，這裡簡單講述一下，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 Con sumer 配置、效能調優及監控

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 執行緒；

如何保證資料不丟

這個是常用的配置，

block.on.buffer.full ：預設設定為 false，當達到記憶體設定時，可能通過 block 停止接受新的 record 或者丟擲一些錯誤，預設情況下，Producer 將不會丟擲 BufferExhaustException，而是當達到 max.block.ms 這個時間後直接丟擲 TimeoutException。設定為 true 的意義就是將 max.block.ms 設定為 Long.MAX_VALUE，未來版本中這個設定將被遺棄，推薦設定 max.block.ms 。