歡迎大家前往雲+社區，獲取更多騰訊海量技術實踐干貨哦~
作者：mikealzhou
本文重點介紹kafka的兩類常見數據遷移方式：1、broker內部不同數據盤之間的分區數據遷移；2、不同broker之間的分區數據遷移。
一、broker 內部不同數據盤之間進行分區數據遷移
1.1 背景介紹
最近，騰訊雲的一個重要客戶發現kafka broker內部的topic分區數據存儲分布不均勻，導致部分磁盤100%耗盡，而部分磁盤只有40%的消耗量。
分析原因，發現存在部分topic的分區數據過於集中在某些磁盤導致，比如，以下截圖顯示的/data5 數據盤。
根據分布式系統的特點，很容易想到采取數據遷移的辦法，對broker內部不同數據盤的分區數據進行遷移。在進行線上集群數據遷移之前，為了保證生產集群的數據完整和安全，必須先在測試集群進行測試。
1.2 測試broker內部不同數據盤進行分區數據遷移
1.2.1 建立測試topic並驗證生產和消費正常
我們搭建的測試集群，Kafka 有三個broker，hostname分別為：tbds-172-16-16-11，tbds-172-16-16-12，tbds-172-16-16-16。每個broker配置了兩塊數據盤，緩存數據分別存儲在 /data/kafka-logs/ 和 /data1/kafka-logs/。
首先建立測試topic：
./kafka-topics.sh --create --zookeeper tbds-172-16-16-11:2181 --replication-factor 2 --partitions 3 --topic test_topic
然后向topic生產發送500條數據，發送的時候也同時消費數據。然后查看topic的分區數據情況：
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG OWNER groupid1 test_topic 0 172 172 0 kafka-python-1.3.1_tbds-172-16-16-3/172.16.16.3 groupid1 test_topic 1 156 156 0 kafka-python-1.3.1_tbds-172-16-16-3/172.16.16.3 groupid1 test_topic 2 172 172 0 kafka-python-1.3.1_tbds-172-16-16-3/172.16.16.3
發現test_topic生產和消費數據都正常。
1.2.2 將分區數據在磁盤間進行遷移
現在登錄tbds-172-16-16-12這台broker節點，將test_topic的分區數據目錄 /data1/kafka-logs/test_topic-0/ 移動到 /data/kafka-logs/ ：
mv /data1/kafka-logs/test_topic-0/ /data/kafka-logs/
查看 /data/kafka-logs/ 目錄下，分區test_topic-0 的數據：
1.2.3 再次對測試topic生產和消費數據
再次發送500條數據，同時消費數據。然后查看數據情況：
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG OWNER groupid1 test_topic 0 337 337 0 kafka-python-1.3.1_tbds-172-16-16-3/172.16.16.3 groupid1 test_topic 1 304 304 0 kafka-python-1.3.1_tbds-172-16-16-3/172.16.16.3 groupid1 test_topic 2 359 359 0 kafka-python-1.3.1_tbds-172-16-16-3/172.16.16.3
再次查看tbds-172-16-16-12 這個broker節點的/data/kafka-logs/test_topic-0/ 分區目錄下的數據：
發現，從 /data1/kafka-logs/ 移動到 /data/kafka-logs/ 目錄下的分區數據目錄test_topic-0/（也就是編號為0的分區）緩存數據並沒有增加。
因為test_topic每個分區有2個replicas，因此，我找到編號為0的另外一個分區replica數據存儲在tbds-172-16-16-16這台broker節點。登錄tbds-172-16-16-16這個broker節點，打開編號為0的分區緩存數據目錄，得到如下信息：
發現，tbds-172-16-16-16這台broker節點的分區數據目錄test_topic-0/內緩存數據量是增加的，也就是緩存有再次生產發送的message數據。
由此可見，經過移動之后的tbds-172-16-16-12這台broker節點的編號為0的分區數據緩存目錄內，並沒有新增緩存數據。與之對應的，沒有做分區數據移動操作的 tbds-172-16-16-16這台broker 節點的編號為0的分區緩存數據目錄內新增再次發送的數據。
是不是意味着不能在broker的磁盤間移動分區數據呢？
1.2.4 調用重啟大法：重啟kafka
重啟kafka集群，重啟完成后，發現tbds-172-16-16-12這台broker節點的編號為0的分區緩存數據目錄內的數據也增加到正常水平。
表明重啟之后，broker的不同磁盤間遷移數據已經生效。
1.2.5 驗證磁盤間遷移分區數據生效
再次向test_topic發送500條數據，同時消費數據，然后查看數據情況：
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG OWNER groupid1 test_topic 0 521 521 0 kafka-python-1.3.1_tbds-172-16-16-3/172.16.16.3 groupid1 test_topic 1 468 468 0 kafka-python-1.3.1_tbds-172-16-16-3/172.16.16.3 groupid1 test_topic 2 511 511 0 kafka-python-1.3.1_tbds-172-16-16-3/172.16.16.3
查看tbds-172-16-16-12 和 tbds-172-16-16-16 兩個broker節點的test_topic-0分區數據的緩存目錄：
發現兩個replicas完全一樣。
1.3 結論
Kafka broker 內部不同數據盤之間可以自由遷移分區數據目錄。遷移完成后，重啟kafka即可生效。
二、不同broker之間傳輸分區數據
當對kafka集群進行擴容之后，由於新擴容的broker沒有緩存數據，容易造成系統的數據分布不均勻。因此，需要將原來集群broker的分區數據遷移到新擴容的broker節點。
不同broker之間傳輸分區數據，可以使用kafka自帶的kafka-reassign-partitions.sh腳本工具實現。
我們在kafka測試集群原有的3台broker基礎上，擴容1台broker。
2.1 獲取test_topic的分區分布情況
執行命令：
./kafka-topics.sh --zookeeper 172.16.16.11:2181 --topic test_topic --describe
可以得到test_topic的3個分區（每個分區有2份replicas）在三個broker節點的分布情況：
Topic:test_topic PartitionCount:3 ReplicationFactor:2 Configs: Topic: test_topic Partition: 0 Leader: 1002 Replicas: 1002,1001 Isr: 1002,1001 Topic: test_topic Partition: 1 Leader: 1003 Replicas: 1003,1002 Isr: 1003,1002 Topic: test_topic Partition: 2 Leader: 1001 Replicas: 1001,1003 Isr: 1001,1003
2.2 獲取topic重新分區的配額文件
編寫分配腳本：move_kafka_topic.json內容如下：
{"topics": [{"topic":"test_topic"}], "version": 1}
執行分配計划生成腳本：
./kafka-reassign-partitions.sh --zookeeper tbds-172-16-16-11:2181 --topics-to-move-json-file /tmp/move_kafka_topic.json --broker-list "1001,1002,1003,1004" --generate
命令里面的broker-list填寫kafka集群4個broker的id。不同kafka集群，因為部署方式不一樣，選擇的broker id也不一樣。我們的測試集群broker id是1001,1002,1003,1004。讀者需要根據自己的kafka集群設置的broker id填寫。
執行命令之后，得到以下結果：
Current partition replica assignment #當前分區的副本分配
{"version":1,"partitions":[{"topic":"test_topic","partition":0,"replicas":[1002,1001]},{"topic":"test_topic","partition":2,"replicas":[1001,1003]},{"topic":"test_topic","partition":1,"replicas":[1003,1002]}]} Proposed partition reassignment configuration #建議的分區配置 {"version":1,"partitions":[{"topic":"test_topic","partition":0,"replicas":[1001,1002]},{"topic":"test_topic","partition":2,"replicas":[1003,1004]},{"topic":"test_topic","partition":1,"replicas":[1002,1003]}]}
Proposed partition reassignment configuration 后是根據命令行的指定的broker list生成的分區分配計划json格式。將 Proposed partition reassignment configuration的配置復制保存到一個文件中 move_kafka_topic_result.json：
{"version":1,"partitions":[{"topic":"test_topic","partition":0,"replicas":[1001,1002]},{"topic":"test_topic","partition":2,"replicas":[1003,1004]},{"topic":"test_topic","partition":1,"replicas":[1002,1003]}]}
2.3 對topic分區數據進行重新分布
執行重新分配命令：
./kafka-reassign-partitions.sh --zookeeper tbds-172-16-16-11:2181 --reassignment-json-file /tmp/move_kafka_topic_result.json --execute
得到如下結果：
Current partition replica assignment
{"version":1,"partitions":[{"topic":"test_topic","partition":0,"replicas":[1002,1001]},{"topic":"test_topic","partition":2,"replicas":[1001,1003]},{"topic":"test_topic","partition":1,"replicas":[1003,1002]}]} Save this to use as the --reassignment-json-file option during rollback Successfully started reassignment of partitions {"version":1,"partitions":[{"topic":"test_topic","partition":0,"replicas":[1001,1002]},{"topic":"test_topic","partition":2,"replicas":[1003,1004]},{"topic":"test_topic","partition":1,"replicas":[1002,1003]}]}
從返回結果來看，分區數據重新分布任務已經啟動成功。
2.4 查看分區數據重新分布進度
檢查分配的狀態，執行命令：
./kafka-reassign-partitions.sh --zookeeper tbds-172-16-16-11:2181 --reassignment-json-file /tmp/move_kafka_topic_result.json --verify
得到結果：
Status of partition reassignment: Reassignment of partition [test_topic,0] completed successfully Reassignment of partition [test_topic,2] completed successfully Reassignment of partition [test_topic,1] completed successfully
表明分區數據重新分步任務已經完成。
2.5 再次獲取test_topic的分區分布情況
再次查看各個分區的分布情況，執行命令：
./kafka-topics.sh --zookeeper 172.16.16.11:2181 --topic test_topic --describe
得到返回結果：
Topic:test_topic PartitionCount:3 ReplicationFactor:2 Configs: Topic: test_topic Partition: 0 Leader: 1002 Replicas: 1001,1002 Isr: 1002,1001 Topic: test_topic Partition: 1 Leader: 1003 Replicas: 1002,1003 Isr: 1003,1002 Topic: test_topic Partition: 2 Leader: 1003 Replicas: 1003,1004 Isr: 1003,1004
從結果看出，test_topic的分區數據已經由原來的3個broker，重新分布到4個broker。
三、測試結論
Ø Kafka broker 內部不同數據盤之間可以自由遷移分區數據目錄。遷移完成后，重啟kafka即可生效；
Ø Kafka 不同broker之前可以遷移數據，使用kafka自帶的kafka-reassign-partitions.sh腳本工具實現。
四、修復客戶的kafka集群故障
我們采用本文測試的方法，對該客戶的Kafka集群進行broker節點內部不同磁盤間的數據遷移，對多個topic均進行了數據遷移，最終實現磁盤間的數據緩存分布均勻化。
同時，我們又對客戶的kafka集群進行擴容，擴容之后采用本文描述的不同broker之間遷移分區數據方法，對多個topic均進行了數據遷移，保證新擴容節點也有緩存數據，原來的broker節點存儲壓力減小。