kafka數據遷移實踐

歡迎你們前往雲+社區，獲取更多騰訊海量技術實踐乾貨哦~

做者：mikealzhou

本文重點介紹kafka的兩類常見數據遷移方式：一、broker內部不一樣數據盤之間的分區數據遷移；二、不一樣broker之間的分區數據遷移。

1、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便可生效。

2、不一樣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。

3、測試結論

Ø Kafka broker 內部不一樣數據盤之間能夠自由遷移分區數據目錄。遷移完成後，重啓kafka便可生效；

Ø Kafka 不一樣broker以前能夠遷移數據，使用kafka自帶的kafka-reassign-partitions.sh腳本工具實現。

4、修復客戶的kafka集羣故障

咱們採用本文測試的方法，對該客戶的Kafka集羣進行broker節點內部不一樣磁盤間的數據遷移，對多個topic均進行了數據遷移，最終實現磁盤間的數據緩存分佈均勻化。

同時，咱們又對客戶的kafka集羣進行擴容，擴容以後採用本文描述的不一樣broker之間遷移分區數據方法，對多個topic均進行了數據遷移，保證新擴容節點也有緩存數據，原來的broker節點存儲壓力減少。

相關閱讀

kafka 運營總結

kafka 設計原理

kubernetes 中 kafka 和 zookeeper 有狀態集羣服務部署實踐 (一)

此文已由做者受權雲加社區發佈，轉載請註明原文出處