避坑指南：Kafka集群快速擴容的方案總結

什么是數(shù)據(jù)遷移

Apache Kafka 對于數(shù)據(jù)遷移的官方說法是分區(qū)重分配。即重新分配分區(qū)在集群的分布情況。官方提供了kafka-reassign-partitions.sh腳本來執(zhí)行分區(qū)重分配操作。其底層實現(xiàn)主要有如下三步：

1. 通過副本復制的機制將老節(jié)點上的分區(qū)搬遷到新的節(jié)點上。

2. 然后再將Leader切換到新的節(jié)點。

3. 最后刪除老節(jié)點上的分區(qū)。

重分配過程中最重要的一步是數(shù)據(jù)復制。故本文用數(shù)據(jù)遷移來形容這一行為，下面來看一下數(shù)據(jù)遷移的過程。

假設topicA有3個分區(qū)，2個副本，分區(qū)和副本分布在節(jié)點1和節(jié)點2。此時加了一個節(jié)點3，如果要讓3個節(jié)點均分壓力，就需要從節(jié)點1，2中遷移兩個分區(qū)到節(jié)點3，如下所示:

圖1：遷移前

圖2：遷移后

擴容場景

了解了數(shù)據(jù)遷移，我們來看下哪些場景需要進行擴容，然后有哪些方法可以實現(xiàn)快速擴容的效果。通常有如下兩種需要緊急擴容的場景：

• 集群所有節(jié)點負載都高，需要快速擴容。

• 集群內某幾臺節(jié)點負載很高，需要降低這些節(jié)點的壓力。

首先，來談談什么是節(jié)點壓力。從我們的運營經(jīng)驗來說，Kafka集群的壓力通常體現(xiàn)在磁盤util、CPU、網(wǎng)卡三個指標上。正常來說，通過加節(jié)點都可以解決這三個指標帶來的問題。但是，從精細化運維的角度來說，可以有針對性地解決負載問題，達到不擴容就可以快速降低集群壓力的目的。比如通過參數(shù)調優(yōu)，踢掉某臺故障節(jié)點/某塊壞盤等等。

關于精細化運維我們在后續(xù)的文章再展開，本文主要是討論如何能通過加節(jié)點實現(xiàn)快速實現(xiàn)集群擴容。

找到Top主題

根據(jù)二八法則和現(xiàn)網(wǎng)運營來看，在大多數(shù)集群中，頭部效應一般都比較明顯，即大部分壓力都是由少量Topic帶來的。所以一般只要解決導致問題的頭部主題，就會事半功倍的解決問題。給大家看一下典型的現(xiàn)網(wǎng)集群的Topic流量排行示意圖，集群的流量集中在下面的Top主題中：

圖3：現(xiàn)網(wǎng)某集群topic流量排序圖

那怎么樣找到Top主題呢？

1. 如果系統(tǒng)內部有通過Broker暴露的的Jmx接口采集Topic入流量指標，那么對這些流量做一個排序，可以快速的找到目標主題。

2. 也可以寫一個shell腳本，使用cmdline-jmxclient.jar工具實時的取到所有topic的流量，然后再做排序。這個方法比較繁瑣，仍需考慮Topic的分區(qū)是否分布在不同的Broker上，是否需要做匯總等。

3. 如果不想用2的辦法，有一個簡單的辦法可以大概看出流量的分布。先進入broker上的數(shù)據(jù)目錄，然后查看每個分區(qū)的堆積的數(shù)據(jù)量大小。比如執(zhí)行如下命令：ll -h /data/kafka_data/。根據(jù)堆積的情況來判斷哪些Topic的流量相對較大。

當然，如果集群中所有主題的流量都非常平均，那就對所有的Topic一起處理。接下來我們來討論下當遇到緊急擴容的需求時，有哪些方案可以選擇。

以下方案的核心思想：遷移盡量少的數(shù)據(jù)，或者不遷移實現(xiàn)壓力的轉移。

方案一：降低數(shù)據(jù)保留時間，精準遷移

這個方案是大家最常用的方法，也是操作起來最簡單的。無論集群整體壓力都高還是某些Broker的壓力大，都可以通過這個方案來執(zhí)行擴容。一般執(zhí)行如下四步：

1. 找出需要操作的Topic

2. 調整這些Topic的數(shù)據(jù)保留時間

3. 對這些Topic進行遷移

4. 等待遷移完成，完成Leader切換，均分集群壓力。

此時當整個集群壓力都很大的時候，這個方案下最好的處理方式是：調整流量最大的分區(qū)的保留時間，然后針對每條broker遷移幾個分區(qū)到新的節(jié)點上。比如，從每臺Broker挑出3個分區(qū)，然后遷移到目標節(jié)點上。如果只有幾臺Broker壓力大，也是一樣的處理方式。

這種方式有兩個缺點：

1. 業(yè)務數(shù)據(jù)的保留時間不一定能調短，因為調短了數(shù)據(jù)保留時間，這些數(shù)據(jù)就會被刪除，無法恢復。所以，當業(yè)務不允許刪除數(shù)據(jù)的時候，這個方案就不允許使用。

2. 當業(yè)務允許調整數(shù)據(jù)保留時間，比如調整到1個小時。此時還有一個小時的數(shù)據(jù)需要遷移，如果此時當前節(jié)點的負載已經(jīng)很高了。此時副本拉取數(shù)據(jù)即會增加當前節(jié)點的負載，導致集群更加無法提供正常服務。當前節(jié)點壓力大的話，可能導致新副本同步數(shù)據(jù)比較慢，會導致集群的壓力沒法快速降下來。

那有沒有方案可以解決這個問題呢？我們再來看下一個方案。

方案二：往指定節(jié)點上添加分區(qū)，均分壓力

如方案一所示，當整個集群壓力都很大時，擴容節(jié)點后，因為數(shù)據(jù)遷移的方案無法使用，新節(jié)點無法承擔壓力，集群負載也降不下來。

此時可以通過擴容分區(qū)到新節(jié)點，將流量導到新的節(jié)點，讓新的節(jié)點也可以承擔流量。一般執(zhí)行如下三步：

• 找出需要操作的Topic

• 評估這些Topic需要導多少流量到新節(jié)點。這一步比較好做，一般比如要把30%，50%的流量導入到新節(jié)點，大概評估一下即可。

• 擴容目標Topic的分區(qū)數(shù)，將這些新增的分區(qū)指定擴容到新的節(jié)點上。

因為擴容Topic的目標分區(qū)是秒級完成的，所以不需要等待。在沒有分區(qū)變更的情況下客戶端的metadata機制默認是每隔30s自動更新一次，所以客戶端會很快感知到分區(qū)變化，默認的生產者寫入策略是輪詢的，此時新增的流量會自動流入到新節(jié)點，原先的節(jié)點的負載很快就降下來了。

方案二的核心點：新增擴容分區(qū)，比如指定添加到目標節(jié)點。如果只是擴容分區(qū)，而這些分區(qū)還是落到老的節(jié)點上，是解決不了問題的。因為一般情況下Topic的分區(qū)的流量都是均勻的，假設Topic當前分區(qū)是100，想讓新的節(jié)點承擔該Topic 50%的壓力，可以將該Topic的分區(qū)數(shù)擴容到200。如下圖所示：

這種方式有如下幾個缺點：

1. 如果業(yè)務邏輯限制了分區(qū)數(shù)不可變更，比如業(yè)務根據(jù)分區(qū)數(shù)做了哈希，根據(jù)哈希值往分區(qū)寫數(shù)據(jù)，又需要要求數(shù)據(jù)保持有序。那么該方案就行不通了。但是一般情況下，有如上業(yè)務場景的Topic數(shù)據(jù)量都比較小，不會成為瓶頸Topic。反之，成為瓶頸的Topic，數(shù)據(jù)量都很大，一般都允許擴容分區(qū)。所以，這點通常情況下不會成為限制，但是在操作前最好和業(yè)務確認下。

3. 如果客戶端寫入有傾斜，即客戶端指定了寫入策略，只針對某些分區(qū)寫入，現(xiàn)在的方案就會失效。遇到這種情況，只能協(xié)調客戶端處理。

雖然這個方案有以上幾個缺點，但是整體方案可以應對90%以上的緊急情況。所以，在不用遷移就降低集群負載的情況下，這個方案是很好用的。

方案三：切換Leader，降低單機負載

方案二比較適合整個集群負載較高的場景，或者因為某些頭部Topic的流量均勻的集中在部分節(jié)點的情況下。現(xiàn)在我們來看一下如下場景：

假設集群有20臺節(jié)點，節(jié)點的機型和規(guī)格是一樣的，從網(wǎng)卡進出流量上來看，流量是均衡的。但是因為節(jié)點間性能的差異(原因可能是機型年限不一樣，不同節(jié)點上Topic的業(yè)務形態(tài)不一樣等等)，導致某幾臺節(jié)點的負載很高。

此時，如果通過方案二處理，就會有種殺雞用牛刀的感覺，而且效果并不好。為什么呢？

假設頭部流量的Topic分區(qū)均勻分配在了20臺節(jié)點上。如果要通過擴容分區(qū)降低其中幾臺的負載，因為生產端是均勻寫入的，則需要擴容很多倍的分區(qū)到新的節(jié)點上，才能把這幾臺的流量降下來。這會導致該Topic的分區(qū)數(shù)急速膨脹。

在這種情況下，我們可以嘗試切換分區(qū) Leader的方案來降低單機的壓力。這個操作比較簡單，因為Leader切換的速度是秒級的，所以見效也很快。思路如下：

1. 找出負載高的節(jié)點上的所有Leader

2. 將節(jié)點上分區(qū)的Leader切換到負載較低的Follower節(jié)點上

  圖5：切換Leader前

  圖6：切換Leader后

• 這種方式適用于部分節(jié)點負載較高的情況，因為負載會轉移到Follower上，如果Follower的負載本身就很高，則這種手段會加大Follower的壓力。

• 當Leader出現(xiàn)負載較高的時候，副本可能會掉出ISR。因為Leader負載高，F(xiàn)ollower拉取數(shù)據(jù)慢，導致副本跟不上Leader，掉出ISR。當出現(xiàn)這種情況，就不能切換Leader，強行切換的話，會出現(xiàn)數(shù)據(jù)截斷，導致數(shù)據(jù)丟失。

我們繼續(xù)來看一下，有沒有其他方法可以解決這種問題。

方案四：單副本運行，降低單機負載

當出現(xiàn)方案三中的無法處理的情況時，即無法使用切換Leader的手段降低壓力時。我們可以通過將高負載節(jié)點上的分區(qū)Follower剔除，將分區(qū)切換為單副本運行，來臨時降低節(jié)點的壓力，讓集群暫時的快速回歸正常。

1. 找出負載高節(jié)點上的所有Follower

2. 將節(jié)點上的Follower暫時移除

這個操作的原理：分區(qū)的Follower會不斷的從Leader同步數(shù)據(jù)，即從Leader拉取數(shù)據(jù)到本地進行報錯。這個行為會占用CPU，網(wǎng)卡，磁盤等資源，如果把這一部分流量去掉，則會把節(jié)點上這部分負載空出來。能快速地降低節(jié)點壓力。

圖7: 切掉Follower前

圖8: 切掉Follower后

這種方式的缺點：切掉Follower后，分區(qū)就是單副本運行，沒有備份。如果這臺節(jié)點掛了，那服務就會出現(xiàn)異常，可能出現(xiàn)數(shù)據(jù)寫入異常。

這個步驟還有一個執(zhí)行風險：切掉Follower，節(jié)點負載降低后，一定要記得把停掉分區(qū)分區(qū)的Follower重新啟動，否則單副本運行的數(shù)據(jù)，沒有備份，很容易出現(xiàn)故障。

圖9:  在其他節(jié)點拉起Follower

方案五：其他思路

除了上面提到的擴容方案。還有其他一些可選的解決方法。這些方案的缺點都比較明顯，比如可能出現(xiàn)數(shù)據(jù)丟失，業(yè)務中斷，或者需要客戶端配合，但勝在見效快。

選擇這種方案需要從業(yè)務來考慮：為了快速恢復業(yè)務，可以允許一定程度的數(shù)據(jù)丟失和服務中斷。所以方案還是存在一定的風險，需要和業(yè)務側討論后，再決定可否執(zhí)行。一起簡單的來看下方案思路及其缺點：

1. 刪除Topic重建

因為創(chuàng)建Topic的時候，Kafka默認的算法會將分區(qū)均勻的放到所有節(jié)點上。所以可以通過刪除，重建Topic的形式快速擴容，均分壓力。操作步驟如下：

1. 往集群添加新節(jié)點

2. 刪除流量大的Topic

3. 重建Topic

當Topic創(chuàng)建完成后，流量就會均勻寫入到所有節(jié)點。整個過程最大的優(yōu)點就是恢復快，幾分鐘就完成了。缺點就是：

1. 如果刪除Topic前，消費還有堆積，則這些數(shù)據(jù)就不會被消費到，會丟失。

2. 在重建Topic的過程，客戶端會報UnknownTopicOrPartitionException錯誤

2. 垂直升配，替換為更高規(guī)格的節(jié)點

如果集群負載的壓力是在Cpu或者網(wǎng)卡，并且是使用云上的虛擬機搭建的kafka集群，可以利用CVM的熱遷移能力，垂直升配虛擬機的規(guī)格。

這個方法也是大家通常的做法。但是在做這個操作的時候，有一個注意點，需要關注這個節(jié)點上是否有未同步副本。也需要關注一下集群參數(shù)unclean.leader.election.enable 參數(shù)的值。

如果存在未同步副本，當 unclean.leader.election.enable=true 時，則表示允許選擇不再ISR中的副本為Leader。此時如果垂直升配，則會出現(xiàn)未同步副本當選為Leader，出現(xiàn)數(shù)據(jù)階段，出現(xiàn)數(shù)據(jù)丟失。

如果存在未同步副本，當 unclean.leader.election.enable=false 時，則表示不允許選擇不再ISR中的副本為Leader。此時在CVM升配過程，Broker重啟的過程中，就會出現(xiàn)服務中斷。但是不會數(shù)據(jù)截斷導致的數(shù)據(jù)丟失。

3.客戶端控制寫入分區(qū)策略

這種方法嚴重依賴客戶端。需要客戶端有指定往哪些分區(qū)寫入數(shù)據(jù)的能力。因為大部分業(yè)務的Producer客戶端，基本都是直接調用官方SDK的Produce方法進行數(shù)據(jù)發(fā)送，數(shù)據(jù)會均勻的寫入到所有分區(qū)。所以大部分客戶端沒有這個能力。而一旦業(yè)務的客戶端可以動態(tài)的指定分區(qū)寫入數(shù)據(jù)(官方SDK自帶指定分區(qū)寫入的功能)。降低負載就會變的很簡單，不需要進行數(shù)據(jù)遷移。

即讓客戶端指定數(shù)據(jù)寫入到負載較低的分區(qū)，就可以降低高負載節(jié)點的壓力。

4.從最新的數(shù)據(jù)遷移，丟棄老數(shù)據(jù)

這個思路是之前網(wǎng)上看到的一個方案，適用于某些允許數(shù)據(jù)丟失的場景。因為Kakfa默認的遷移機制，新副本都是從分區(qū)最早的可用的位置拉取數(shù)據(jù)，然后進行同步。

在某些場景，如數(shù)據(jù)量很大，數(shù)據(jù)保留時間卻很短的場景中，可能出現(xiàn)當副本同步完數(shù)據(jù)后，這些數(shù)據(jù)其實已經(jīng)過期了，同步完數(shù)據(jù)就需要立即刪除。則這個遷移的行為即是多余的。所以還不如直接從最新的數(shù)據(jù)進行同步，這樣在新添加的節(jié)點上的新副本立即就可以加入ISR。則立即可以Leader提供服務。

這個方案的缺點是：

1. 如果新加入的副本立即進入ISR，同時又成為Leader，則會出現(xiàn)數(shù)據(jù)丟失，即從加入ISR那一刻起，往前推這個Topic的數(shù)據(jù)保留時間這段時間的數(shù)據(jù)都會丟失。

2. 需要修改Broker默認的數(shù)據(jù)復制機制，對研發(fā)能力要求較高。

關于架構的一些想法

看到這里，會發(fā)現(xiàn)整體下來，Kafka的擴容還是不夠靈活，快捷和方便。操作起來比較麻煩。有沒有更好的方案呢？

首先來看下造成擴容問題的原因，是受Kafka本身架構的限制。Kafka 是以分區(qū)為讀寫單位，分區(qū)是和節(jié)點綁定的，這些數(shù)據(jù)會寫入到元數(shù)據(jù)存儲中。此時一旦計算層(CPU/網(wǎng)卡)或存儲層(util)出現(xiàn)瓶頸，是沒辦法讓其他節(jié)點承載壓力的。如果要解決這個問題，Kafka在架構上要做很大的改動。

從架構的角度出發(fā)，我個人理解，解決的思路就是：計算存儲分離 + 存儲分段。這一點Apache Pulsar就做的很好。我們來簡單看一下Pulsar的做法。來看下圖：

圖10：Pulsar 簡要架構圖

計算存儲分離：解決的是計算壓力的快速轉移。計算節(jié)點和存儲節(jié)點是分開的。計算節(jié)點只負責計算邏輯的處理，是無狀態(tài)的節(jié)點。當節(jié)點出現(xiàn)瓶頸，可以快速橫向擴容。

存儲分段：解決的主要是存儲層IO壓力的快速轉移。Pulsar使用Bookeeper作為存儲層，Pulsar將邏輯上的分區(qū)，在實際存儲層面，分為多個段(segment)進行管理和存儲。如果出現(xiàn)某個存儲的機器有瓶頸，直接禁用該機器上segment，在新的機器上拉起新的Segment即可。

總結一下，一旦Pulsar集群遇到上面說的Kafka集群類似的瓶頸，從擴容的角度來說，會更優(yōu)雅和便捷。這是架構自身帶來的優(yōu)勢。

總結

本文列舉了一些快速擴容的手段和方案，幫助大家盡量避免遷移、盡量降低需要遷移的數(shù)據(jù)量。大家遇到問題的時候，可以根據(jù)實際的業(yè)務場景選擇適合的方案進行處理。

Apache Kafka 的擴容復雜度源于Kakfa存算一體的架構。即數(shù)據(jù)生產和消費都是以分區(qū)為單位的，而分區(qū)從創(chuàng)建開始就會和某一個節(jié)點進行綁定。如果沒有進行分區(qū)遷移，則分區(qū)和節(jié)點的綁定關系不會發(fā)生改變。當遇到節(jié)點出現(xiàn)性能問題時，這個分區(qū)也會受到影響，從而產生本文討論的問題。

在云原生架構，存算分離成為了一種趨勢。在消息隊列領域，最近新興的開源消息隊列Apache Pulsar在架構上實現(xiàn)了存儲計算分離，通過Apache Bookeeper 實現(xiàn)對分區(qū)數(shù)據(jù)的分段分節(jié)點存儲，從而避免了Kakfa遇到的擴容遷移問題。