Kafka資料遷移
1.概述
Kafka的使用場景非常廣泛，一些實時流資料業務場景，均依賴Kafka來做資料分流。而在分散式應用場景中，資料遷移是一個比較常見的問題。關於Kafka叢集資料如何遷移，今天筆者將為大家詳細介紹。
2.內容
本篇部落格為大家介紹兩種遷移場景，分別是同叢集資料遷移、跨叢集資料遷移。如下圖所示：
2.1 同叢集遷移
同叢集之間資料遷移，比如在已有的叢集中新增了一個Broker節點，此時需要將原來叢集中已有的Topic的資料遷移部分到新的叢集中，緩解叢集壓力。
將新的節點新增到Kafka叢集很簡單，只需為它們分配一個唯一的Broker ID，並在新伺服器上啟動Kafka。但是，這些新伺服器節點不會自動分配任何資料分割槽，因此除非將分割槽移動到新增的節點，否則在建立新Topic之前新節點不會執行任何操作。因此，通常在將新伺服器節點新增到Kafka叢集時，需要將一些現有資料遷移到這些新的節點。
遷移資料的過程是手動啟動的，執行過程是完全自動化的。在Kafka後臺服務中，Kafka將新增新伺服器作為其正在遷移的分割槽的Follower，並允許新增節點完全複製該分割槽中的現有資料。當新伺服器節點完全複製此分割槽的內容並加入同步副本（ISR）時，其中一個現有副本將刪除其分割槽的資料。
Kafka系統提供了一個分割槽重新分配工具（kafka-reassign-partitions.sh），該工具可用於在Broker之間遷移分割槽。理想情況下，將確保所有Broker的資料和分割槽均勻分配。分割槽重新分配工具無法自動分析Kafka群集中的資料分佈並遷移分割槽以實現均勻的負載均衡。因此，管理員在操作的時候，必須弄清楚應該遷移哪些Topic或分割槽。
分割槽重新分配工具可以在3種互斥模式下執行：
- --generate：在此模式下，給定Topic列表和Broker列表，該工具會生成候選重新分配，以將指定Topic的所有分割槽遷移到新Broker中。此選項僅提供了一種方便的方法，可在給定Topic和目標Broker列表的情況下生成分割槽重新分配計劃。
- --execute：在此模式下，該工具將根據使用者提供的重新分配計劃啟動分割槽的重新分配。 （使用--reassignment-json-file選項）。由管理員手動制定自定義重新分配計劃，也可以使用--generate選項提供。
- --verify：在此模式下，該工具將驗證最後一次--execute期間列出的所有分割槽的重新分配狀態。狀態可以有成功、失敗或正在進行等狀態。
2.1.1 遷移過程實現
分割槽重新分配工具可用於將一些Topic從當前的Broker節點中遷移到新新增的Broker中。這在擴充套件現有叢集時通常很有用，因為將整個Topic移動到新的Broker變得更容易，而不是一次移動一個分割槽。當執行此操作時，使用者需要提供已有的Broker節點的Topic列表，以及到新節點的Broker列表（源Broker到新Broker的對映關係）。然後，該工具在新的Broker中均勻分配給指定Topic列表的所有分割槽。在遷移過程中，Topic的複製因子保持不變。
現有如下例項，將Topic為ke01，ke02的所有分割槽從Broker1中移動到新增的Broker2和Broker3中。由於該工具接受Topic的輸入列表作為JSON檔案，因此需要明確遷移的Topic並建立json檔案，如下所示：
> cat topic-to-move.json
{"topics": [{"topic": "ke01"},
{"topic": "ke02"}],
"version":1
}
準備好JSON檔案，然後使用分割槽重新分配工具生成候選分配，命令如下：
> bin/kafka-reassign-partitions.sh --zookeeper dn1:2181 --topics-to-move-json-file topics-to-move.json --broker-list "1,2" --generate
執行命名之前，Topic（ke01、ke02）的分割槽如下圖所示：
執行完成命令之後，控制檯出現如下資訊：
該工具生成一個候選分配，將所有分割槽從Topic ke01，ke02移動到Broker1和Broker2。需求注意的是，此時分割槽移動尚未開始，它只是告訴你當前的分配和建議。儲存當前分配，以防你想要回滾它。新的賦值應儲存在JSON檔案（例如expand-cluster-reassignment.json）中，以使用--execute選項執行。JSON檔案如下：
{"version":1,"partitions":[{"topic":"ke02","partition":0,"replicas":[2]},{"topic":"ke02","partition":1,"replicas":[1]},{"topic":"ke02","partition":2,"replicas":[2]},{"topic":"ke01","partition":0,"replicas":[2]},{"topic":"ke01","partition":1,"replicas":[1]},{"topic":"ke01","partition":2,"replicas":[2]}]}
執行命令如下所示：
> ./kafka-reassign-partitions.sh --zookeeper dn1:2181 --reassignment-json-file expand-cluster-reassignment.json --execute
最後，--verify選項可與該工具一起使用，以檢查分割槽重新分配的狀態。需要注意的是，相同的expand-cluster-reassignment.json（與--execute選項一起使用）應與--verify選項一起使用，執行命令如下：
> ./kafka-reassign-partitions.sh --zookeeper dn1:2181 --reassignment-json-file expand-cluster-reassignment.json --verify
執行結果如下圖所示：
同時，我們可以通過 Kafka Eagle 工具來檢視Topic的分割槽情況。
2.2 跨叢集遷移
這裡跨叢集遷移，我們指的是在Kafka多個叢集之間複製資料“映象”的過程，以避免與單個叢集中的節點之間發生的複製混淆。 Kafka附帶了一個用於在Kafka叢集之間映象資料的工具。該工具從源叢集使用並生成到目標叢集。這種映象的一個常見用例是在另一個數據中心提供副本。
另外，你可以執行許多此類映象程序以提高吞吐量和容錯（如果一個程序終止，其他程序將佔用額外負載）。將從源叢集中的Topic讀取資料，並將其寫入目標叢集中具有相同名稱的主題。事實上，“映象”資料只不過是一個Kafka將消費者和生產者聯絡在了一起。
源叢集和目標叢集是完全獨立的實體，它們可以具有不同數量的分割槽，並且偏移量將不相同。出於這個原因，映象叢集並不是真正意圖作為容錯機制（因為消費者的位置會有所不同）;為此，建議使用正常的叢集內複製。但是，映象程序將保留並使用訊息Key進行分割槽，因此可以按Key保留順序。
下面是一個跨叢集的單Topic例項，命令如下：
> ./kafka-mirror-maker.sh --consumer.config consumer.properties --producer.config producer.properties --whitelist ke03
需要注意的是，consumer.properties檔案配置源Kafka叢集Broker地址，producer.properties檔案配置目標Kafka叢集地址。如果需要遷移多個Topic，可以使用 --whitelist 'A|B'，如果需要遷移所有的Topic，可以使用 --whitelist '*'。
3.結果預覽
執行跨叢集遷移命令後，目標叢集中使用Kafka Eagle中檢視Topic Size大小看是否與源叢集的Topic Size大小相等，或者使用SQL語句，驗證是否有資料遷移過來，結果如下圖所示：
4.總結
跨叢集遷移資料的本質是，Kafka啟動了消費者讀取源叢集資料，並將消費後的資料寫入到目標叢集，在遷移的過程中，可以啟動多個例項，提供遷出的吞吐量。
5.結束語
這篇部落格就和大家分享到這裡，如果大家在研究學習的過程當中有什麼問題，可以加群進行討論或傳送郵件給我，我會盡我所能為您解答，與君共勉！
另外，博主出書了《 Kafka並不難學 》，喜歡的朋友或同學， 可以在公告欄那裡點選購買連結購買博主的書進行學習，在此感謝大家的支援。