[[420468]]

本篇文章講解以下內容

【kafka運維】副本擴縮容、數據遷移、副本重分配、副本跨路徑遷移

腳本參數

 1. 腳本的使用介紹

該腳本是kafka提供用來重新分配分區的腳本工具;

1.1 生成推薦配置腳本

關鍵參數--generate

在進行分區副本重分配之前,最好是用下面方式獲取一個合理的分配文件; 編寫move-json-file.json文件; 這個文件就是告知想對哪些Topic進行重新分配的計算

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

然后執行下面的腳本,--broker-list "0,1,2,3" 這個參數是你想要分配的Brokers;

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

執行完畢之后會打印

Current partition replica assignment//當前副本分配方式 
{"version":1,"partitions":[{"topic":"test_create_topic1","partition":2,"replicas":[1],"log_dirs":["any"]},{"topic":"test_create_topic1","partition":1,"replicas":[3],"log_dirs":["any"]},{"topic":"test_create_topic1","partition":0,"replicas":[2],"log_dirs":["any"]}]} 
 
Proposed partition reassignment configuration//期望的重新分配方式 
{"version":1,"partitions":[{"topic":"test_create_topic1","partition":2,"replicas":[2],"log_dirs":["any"]},{"topic":"test_create_topic1","partition":1,"replicas":[1],"log_dirs":["any"]},{"topic":"test_create_topic1","partition":0,"replicas":[0],"log_dirs":["any"]}]} 

需求注意的是，此時分區移動尚未開始，它只是告訴你當前的分配和建議。保存當前分配，以防你想要回滾它

1.2. 執行Json文件

關鍵參數--execute 將上面得到期望的重新分配方式文件保存在一個json文件里面 reassignment-json-file.json

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

然后執行

sh bin/kafka-reassign-partitions.sh --zookeeper xxxxx:2181 --reassignment-json-file config/reassignment-json-file.json --execute 

遷移過程注意流量陡增對集群的影響 Kafka提供一個broker之間復制傳輸的流量限制，限制了副本從機器到另一臺機器的帶寬上限，當重新平衡集群，引導新broker，添加或移除broker時候，這是很有用的。因為它限制了這些密集型的數據操作從而保障了對用戶的影響、 例如我們上面的遷移操作加一個限流選項-- throttle 50000000

> sh bin/kafka-reassign-partitions.sh --zookeeper xxxxx:2181 --reassignment-json-file config/reassignment-json-file.json --execute -- throttle 50000000 

在后面加上一個—throttle 50000000 參數, 那么執行移動分區的時候,會被限制流量在50000000 B/s 加上參數后你可以看到

The throttle limit was set to 50000000 B/s 
Successfully started reassignment of partitions. 

需要注意的是,如果你遷移的時候包含 副本跨路徑遷移(同一個Broker多個路徑)那么這個限流措施不會生效,你需要再加上|--replica-alter-log-dirs-throttle 這個限流參數,它限制的是同一個Broker不同路徑直接遷移的限流;

如果你想在重新平衡期間修改限制，增加吞吐量，以便完成的更快。你可以重新運行execute命令，用相同的reassignment-json-file

1.3. 驗證

關鍵參數--verify 該選項用于檢查分區重新分配的狀態，同時—throttle流量限制也會被移除掉; 否則可能會導致定期復制操作的流量也受到限制。

sh bin/kafka-reassign-partitions.sh --zookeeper xxxx:2181 --reassignment-json-file config/reassignment-json-file.json --verify圖片注意: 當你輸入的BrokerId不存在時,該副本的操作會失敗,但是不會影響其他的;例如圖片

2. 副本擴縮

kafka并沒有提供一個專門的腳本來支持副本的擴縮, 不像kafka-topic.sh腳本一樣,是可以擴分區的; 想要對副本進行擴縮,只能是曲線救國了; 利用kafka-reassign-partitions.sh來重新分配副本

2.1 副本擴容

假設我們當前的情況是 3分區1副本,為了提供可用性,我想把副本數升到2;

2.1.1 計算副本分配方式

我們用步驟1.1的 --generate 獲取一下當前的分配情況,得到如下json

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

我們想把所有分區的副本都變成2,那我們只需修改"replicas": []里面的值了,這里面是Broker列表,排在第一個的是Leader; 所以我們根據自己想要的分配規則修改一下json文件就變成如下

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

注意log_dirs里面的數量要和replicas數量匹配;或者直接把log_dirs選項刪除掉; 這個log_dirs是副本跨路徑遷移時候的絕對路徑

2.1.2 執行--execute

如果你想在重新平衡期間修改限制，增加吞吐量，以便完成的更快。你可以重新運行execute命令，用相同的reassignment-json-file:

2.1.2 驗證--verify

圖片完事之后,副本數量就增加了;

2.2 副本縮容

副本縮容跟擴容是一個意思; 當副本分配少于之前的數量時候,多出來的副本會被刪除; 比如剛剛我新增了一個副本,想重新恢復到一個副本

執行下面的json文件

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

執行之后可以看到其他的副本就被標記為刪除了; 一會就會被清理掉

用這樣一種方式我們雖然是實現了副本的擴縮容, 但是副本的分配需要我們自己來把控好, 要做到負載均衡等等; 那肯定是沒有kafka自動幫我們分配比較合理一點; 那么我們有什么好的方法來幫我們給出一個合理分配的Json文件嗎?PS：

我們之前已經分析過【kafka源碼】創建Topic的時候是如何分區和副本的分配規則 那么我們把這樣一個分配過程也用同樣的規則來分配不就Ok了嗎?--generate本質上也是調用了這個方法,AdminUtils.assignReplicasToBrokers(brokerMetadatas, assignment.size, replicas.size)

具體的實現操作請看 【kafka思考】最小成本的擴縮容副本設計方案

自己寫一個工程來實現類似的方法,如果覺得很麻煩,可以直接使用 LogIKM 的新增副本功能直接幫你做了這個事情;(未來會實現)

3. 分區擴容

kafka的分區擴容是 kafka-topis.sh腳本實現的;不支持縮容 分區擴容請看 【kafka源碼】TopicCommand之alter源碼解析(分區擴容)

4. 分區遷移

分區遷移跟上面同理, 請看 1.1,1.2,1.3 部分;

5. 副本跨路徑遷移

為什么線上Kafka機器各個磁盤間的占用不均勻，經常出現“一邊倒”的情形?這是因為Kafka只保證分區數量在各個磁盤上均勻分布，但它無法知曉每個分區實際占用空間，故很有可能出現某些分區消息數量巨大導致占用大量磁盤空間的情況。在1.1版本之前，用戶對此毫無辦法，因為1.1之前Kafka只支持分區數據在不同broker間的重分配，而無法做到在同一個broker下的不同磁盤間做重分配。1.1版本正式支持副本在不同路徑間的遷移

怎么在一臺Broker上用多個路徑存放分區呢?

只需要在配置上接多個文件夾就行了

############################# Log Basics ############################# 
 
# A comma separated list of directories under which to store log files 
log.dirs=kafka-logs-5,kafka-logs-6,kafka-logs-7,kafka-logs-8 

注意同一個Broker上不同路徑只會存放不同的分區，而不會將副本存放在同一個Broker; 不然那副本就沒有意義了(容災)

怎么針對跨路徑遷移呢?

遷移的json文件有一個參數是log_dirs; 默認請求不傳的話 它是"log_dirs": ["any"] (這個數組的數量要跟副本保持一致) 但是你想實現跨路徑遷移,只需要在這里填入絕對路徑就行了,例如下面

遷移的json文件示例

{ 
  "version": 1, 
  "partitions": [{ 
    "topic": "test_create_topic4", 
    "partition": 2, 
    "replicas": [0], 
    "log_dirs": ["/Users/xxxxx/work/IdeaPj/source/kafka/kafka-logs-5"] 
  }, { 
    "topic": "test_create_topic4", 
    "partition": 1, 
    "replicas": [0], 
    "log_dirs": ["/Users/xxxxx/work/IdeaPj/source/kafka/kafka-logs-6"] 
  }] 
} 

然后執行腳本

sh bin/kafka-reassign-partitions.sh --zookeeper xxxxx --reassignment-json-file config/reassignment-json-file.json --execute --bootstrap-server 
xxxxx:9092 --replica-alter-log-dirs-throttle 10000 

注意 --bootstrap-server 在跨路徑遷移的情況下,必須傳入此參數

如果需要限流的話 加上參數|--replica-alter-log-dirs-throttle ; 跟--throttle不一樣的是 --replica-alter-log-dirs-throttle限制的是Broker內不同路徑的遷移流量;

源碼解析

源碼解析請看文章 【kafka源碼】ReassignPartitionsCommand源碼分析(副本擴縮、數據遷移、分區重分配、副本跨路徑遷移)

視頻鏈接：https://mp.weixin.qq.com/s/fQ03wpctV1dGnmk1r-xEWA

本文轉載自微信公眾號「石臻臻的雜貨鋪」，可以通過以下二維碼關注。轉載本文請聯系石臻臻的雜貨鋪公眾號。