【kafka運(yùn)維】你真的懂?dāng)?shù)據(jù)遷移嗎?(附教學(xué)視頻)

【Kafka運(yùn)維】數(shù)據(jù)遷移、分區(qū)副本重分配、跨路徑遷移、副本擴(kuò)縮容

如果你不想看文章,可以直接看配套的視頻; (

后續(xù)的視頻會(huì)在 公眾號、B站等等各平臺同名號[石臻臻的雜貨鋪]上上傳

)

本期視頻內(nèi)容:

分區(qū)副本重分配教程

重分配的注意事項(xiàng)

LogiKm簡化重分配流程,更高效

跨副本遷移

【Kafka專欄】你真的懂?dāng)?shù)據(jù)遷移嗎？視頻

腳本參數(shù)

1. 腳本的使用介紹

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

1.1 生成推薦配置腳本

關(guān)鍵參數(shù)--generate

在進(jìn)行分區(qū)副本重分配之前,最好是用下面方式獲取一個(gè)合理的分配文件;

編寫move-json-file.json文件; 這個(gè)文件就是告知想對哪些Topic進(jìn)行重新分配的計(jì)算

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

然后執(zhí)行下面的腳本,--broker-list "0,1,2,3" 這個(gè)參數(shù)是你想要分配的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

執(zhí)行完畢之后會(huì)打印

Current partition replica assignment//當(dāng)前副本分配方式 {"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"]}]}

需求注意的是，此時(shí)分區(qū)移動(dòng)尚未開始，它只是告訴你當(dāng)前的分配和建議。保存當(dāng)前分配，以防你想要回滾它

1.2. 執(zhí)行Json文件

關(guān)鍵參數(shù)--execute

將上面得到期望的重新分配方式文件保存在一個(gè)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"]}]}

然后執(zhí)行

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

遷移過程注意流量陡增對集群的影響

Kafka提供一個(gè)broker之間復(fù)制傳輸?shù)牧髁肯拗疲拗屏烁北緩臋C(jī)器到另一臺機(jī)器的帶寬上限，當(dāng)重新平衡集群，引導(dǎo)新broker，添加或移除broker時(shí)候，這是很有用的。因?yàn)樗拗屏诉@些密集型的數(shù)據(jù)操作從而保障了對用戶的影響、

例如我們上面的遷移操作加一個(gè)限流選項(xiàng)-- throttle 50000000

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

在后面加上一個(gè)—throttle 50000000 參數(shù), 那么執(zhí)行移動(dòng)分區(qū)的時(shí)候,會(huì)被限制流量在50000000 B/s

加上參數(shù)后你可以看到

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

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

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

1.3. 驗(yàn)證

關(guān)鍵參數(shù)--verify

該選項(xiàng)用于檢查分區(qū)重新分配的狀態(tài)，同時(shí)—throttle流量限制也會(huì)被移除掉; 否則可能會(huì)導(dǎo)致定期復(fù)制操作的流量也受到限制。

sh bin/kafka-reassign-partitions.sh --zookeeper xxxx:2181 --reassignment-json-file config/reassignment-json-file.json --verify

注意: 當(dāng)你輸入的BrokerId不存在時(shí),該副本的操作會(huì)失敗,但是不會(huì)影響其他的;例如

2. 副本擴(kuò)縮

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

2.1 副本擴(kuò)容

假設(shè)我們當(dāng)前的情況是 3分區(qū)1副本,為了提供可用性,我想把副本數(shù)升到2;

我們用步驟1.1的 --generate 獲取一下當(dāng)前的分配情況,得到如下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"] }] }

我們想把所有分區(qū)的副本都變成2,那我們只需修改"replicas": []里面的值了,這里面是Broker列表,排在第一個(gè)的是Leader; 所以我們根據(jù)自己想要的分配規(guī)則修改一下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里面的數(shù)量要和replicas數(shù)量匹配;或者直接把log_dirs選項(xiàng)刪除掉; 這個(gè)log_dirs是副本跨路徑遷移時(shí)候的絕對路徑

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

完事之后,副本數(shù)量就增加了;

2.2 副本縮容

副本縮容跟擴(kuò)容是一個(gè)意思; 當(dāng)副本分配少于之前的數(shù)量時(shí)候,多出來的副本會(huì)被刪除;

比如剛剛我新增了一個(gè)副本,想重新恢復(fù)到一個(gè)副本

執(zhí)行下面的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"] }] }

執(zhí)行之后可以看到其他的副本就被標(biāo)記為刪除了; 一會(huì)就會(huì)被清理掉

用這樣一種方式我們雖然是實(shí)現(xiàn)了副本的擴(kuò)縮容, 但是副本的分配需要我們自己來把控好, 要做到負(fù)載均衡等等; 那肯定是沒有kafka自動(dòng)幫我們分配比較合理一點(diǎn); 那么我們有什么好的方法來幫我們給出一個(gè)合理分配的Json文件嗎?

PS：

我們之前已經(jīng)分析過【kafka源碼】創(chuàng)建Topic的時(shí)候是如何分區(qū)和副本的分配規(guī)則 那么我們把這樣一個(gè)分配過程也用同樣的規(guī)則來分配不就Ok了嗎？

--generate本質(zhì)上也是調(diào)用了這個(gè)方法,AdminUtils.assignReplicasToBrokers(brokerMetadatas, assignment.size, replicas.size)

具體的實(shí)現(xiàn)操作請看 【kafka思考】最小成本的擴(kuò)縮容副本設(shè)計(jì)方案

自己寫一個(gè)工程來實(shí)現(xiàn)類似的方法,如果覺得很麻煩,可以直接使用

LogIKM 的新增副本功能直接幫你做了這個(gè)事情;(未來會(huì)實(shí)現(xiàn))

3. 分區(qū)擴(kuò)容

kafka的分區(qū)擴(kuò)容是 kafka-topis.sh腳本實(shí)現(xiàn)的;不支持縮容

分區(qū)擴(kuò)容請看 【kafka源碼】TopicCommand之a(chǎn)lter源碼解析(分區(qū)擴(kuò)容)

4. 分區(qū)遷移

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

5. 副本跨路徑遷移

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

怎么在一臺Broker上用多個(gè)路徑存放分區(qū)呢?

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

############################# 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

注意同一個(gè)Broker上不同路徑只會(huì)存放不同的分區(qū)，而不會(huì)將副本存放在同一個(gè)Broker; 不然那副本就沒有意義了(容災(zāi))

怎么針對跨路徑遷移呢？

遷移的json文件有一個(gè)參數(shù)是log_dirs; 默認(rèn)請求不傳的話 它是"log_dirs": ["any"] （這個(gè)數(shù)組的數(shù)量要跟副本保持一致）

但是你想實(shí)現(xiàn)跨路徑遷移,只需要在這里填入絕對路徑就行了,例如下面

遷移的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"] }] }

然后執(zhí)行腳本

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 在跨路徑遷移的情況下,必須傳入此參數(shù)

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

源碼解析

源碼解析請看文章

【kafka源碼】ReassignPartitionsCommand源碼分析(副本擴(kuò)縮、數(shù)據(jù)遷移、分區(qū)重分配、副本跨路徑遷移)

日常運(yùn)維

、

問題排查

=>

滴滴開源LogiKM一站式Kafka監(jiān)控與管控平臺

Kafka 數(shù)據(jù)遷移 視頻

版權(quán)聲明：本文內(nèi)容由網(wǎng)絡(luò)用戶投稿，版權(quán)歸原作者所有，本站不擁有其著作權(quán)，亦不承擔(dān)相應(yīng)法律責(zé)任。如果您發(fā)現(xiàn)本站中有涉嫌抄襲或描述失實(shí)的內(nèi)容，請聯(lián)系我們jiasou666@gmail.com 處理，核實(shí)后本網(wǎng)站將在24小時(shí)內(nèi)刪除侵權(quán)內(nèi)容。

版權(quán)聲明：本文內(nèi)容由網(wǎng)絡(luò)用戶投稿，版權(quán)歸原作者所有，本站不擁有其著作權(quán)，亦不承擔(dān)相應(yīng)法律責(zé)任。如果您發(fā)現(xiàn)本站中有涉嫌抄襲或描述失實(shí)的內(nèi)容，請聯(lián)系我們jiasou666@gmail.com 處理，核實(shí)后本網(wǎng)站將在24小時(shí)內(nèi)刪除侵權(quán)內(nèi)容。