kafka遷移和擴容

概述

將服務器添加到Kafka集群很容易，只需為其分配唯一的代理ID，然后在新服務器上啟動Kafka，但是，不會為這些新服務器自動分配任何資料磁區，因此，除非將磁區移至它們，否則在創建新主題之前它們將不會做任何作業，因此，通常在將計算機添加到群集時，您將需要將一些現有資料遷移到這些計算機，

資料遷移程序是手動啟動的，但完全自動化，在幕后，Kafka會將新服務器添加為要遷移的磁區的跟隨者，并允許其完全復制該磁區中的現有資料，新服務器完全復制該磁區的內容并加入同步副本后，現有副本之一將洗掉其磁區的資料，

磁區重新分配工具可用于在代理之間移動磁區，理想的磁區分配將確保所有代理之間的資料負載和磁區大小均勻，磁區重新分配工具沒有能力自動研究Kafka群集中的資料分布，并四處移動磁區以實作均勻的負載分布，因此，管理員必須弄清楚應該移動哪些主題或磁區，

磁區重新分配工具可以在3種互斥模式下運行：

• generate：在此模式下，給定主題串列和代理串列，該工具會生成候選重新??分配，以將指定主題的所有磁區移至新的代理，給定主題和目標代理的串列，此選項僅提供一種方便的方法來生成磁區重新分配計劃，
• execute：在此模式下，該工具將根據用戶提供的重新分配計劃啟動磁區的重新分配，（使用–reassignment-json-file選項），這可以是管??理員手工制作的自定義重新分配計劃，也可以使用–generate選項提供
• verify：在此模式下，該工具會驗證上一次–execute期間列出的所有磁區的重新分配狀態，狀態可以是成功完成，失敗或進行中

磁區的遷移：

創建一個磁區為三，副本數為二的topic

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 2 --partitions 3 --topic topic_test

查看topic情況

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic topic_test

撰寫遷移檔案

vi topics-to-move.json

topics-to-move.json:

{"topics": [{"topic": "topic_test"}], "version":1}

一旦json檔案準備就緒，使用磁區重新分配工具生成候選分配

bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --topics-to-move-json-file topics-to-move.json --broker-list "1,2" --generate

Current partition replica assignment
{"version":1,"partitions":[{"topic":"topic_test","partition":2,"replicas":[2,0],"log_dirs":["any","any"]},{"topic":"topic_test","partition":1,"replicas":[1,2],"log_dirs":["any","any"]},{"topic":"topic_test","partition":0,"replicas":[0,1],"log_dirs":["any","any"]}]}

Proposed partition reassignment configuration
{"version":1,"partitions":[{"topic":"topic_test","partition":0,"replicas":[2,1],"log_dirs":["any","any"]},{"topic":"topic_test","partition":2,"replicas":[2,1],"log_dirs":["any","any"]},{"topic":"topic_test","partition":1,"replicas":[1,2],"log_dirs":["any","any"]}]}

該工具生成一個候選分配，將所有磁區從主題topic_test移動到代理1,2，但請注意，此時磁區移動尚未開始，它只是告訴您當前的分配和建議的新分配，應保存當前分配，以防您想要回滾它，新的賦值應保
存在json檔案中（例如expand-cluster-reassignment.json），以使用–execute選項輸入到工具
保存當前分配

vi old-expand-cluster-reassignment.json

old-expand-cluster-reassignment.json:

{"version":1,"partitions":[{"topic":"topic_test","partition":2,"replicas":[2,0],"log_dirs":["any","any"]},{"topic":"topic_test","partition":1,"replicas":[1,2],"log_dirs":["any","any"]},{"topic":"topic_test","partition":0,"replicas":[0,1],"log_dirs":["any","any"]}]}

vi expand-cluster-reassignment.json

expand-cluster-reassignment.json:

{"version":1,"partitions":[{"topic":"topic_test","partition":0,"replicas":[2,1],"log_dirs":["any","any"]},{"topic":"topic_test","partition":2,"replicas":[2,1],"log_dirs":["any","any"]},{"topic":"topic_test","partition":1,"replicas":[1,2],"log_dirs":["any","any"]}]}

執行遷移

bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file expand-cluster-reassignment.json --execute

查看遷移結果

可以看到磁區已經遷移成功
也可以通過命令查看結果

bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file expand-cluster-reassignment.json --verify 

副本的遷移

首先我們需要還原原本磁區

bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file old-expand-cluster-reassignment.json --execute

然后將副本2遷移至磁區0中
撰寫遷移檔案

vi increase-replication-factor.json

increase-replication-factor.json:

{"version":1,"partitions":[{"topic":"topic_test","partition":0,"replicas":[0,1,2]}]}

執行遷移

bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file increase-replication-factor.json --execute

可以看到副本2已經成功遷移到磁區1中，