kafka的擴(kuò)容難點(diǎn)：

1)主要在于增加機(jī)器之后，數(shù)據(jù)需要rebalance到新增的空閑節(jié)點(diǎn)，即把partitions遷移到空閑機(jī)器上。

kafka提供了bin/kafka-reassign-partitions.sh工具，完成parttition的遷移。

2)kafka的集群的數(shù)據(jù)量加大，數(shù)據(jù)rebalance的時(shí)間較長(zhǎng)。解決辦法是把log.retention.hours=1設(shè)置一小時(shí)(生產(chǎn)參數(shù)24小時(shí))。

修改參數(shù)之后，重啟kakfa節(jié)點(diǎn)，kafka會(huì)主動(dòng)purge 1小時(shí)之前的log數(shù)據(jù)。

以下是kafka_0.8.1.1版本kafkka集群擴(kuò)容操作記錄，從16臺(tái)物理機(jī)擴(kuò)容到24臺(tái)物理，partition數(shù)量由128個(gè)增加到192個(gè)：

kafka支持參數(shù)的動(dòng)態(tài)修改，下面單獨(dú)設(shè)置某個(gè)topic的retention時(shí)間。

bin/kafka-topics.sh --zookeeper zookeeper-003:2181/kafka --alter --topic order_active --config retention.ms=86400000

1、準(zhǔn)備工作：

1) 驗(yàn)證kafka節(jié)點(diǎn)是否正常加入集群。

zkCli.sh > ls /kafka/brokers/ids/

2)檢查新增機(jī)器的zk信息中主機(jī)名是否準(zhǔn)確，每個(gè)broker都要檢查

zkCli.sh > get /kafka/brokers/ids/1

2、purge數(shù)據(jù)，使數(shù)據(jù)遷移更加快速

1)替換retent.time,只保留最近一個(gè)小時(shí)的數(shù)據(jù)。主要是為了方面topic數(shù)據(jù)快速遷移。

sed -i 's/log.retention.hours=24/log.retention.hours=1/g' /apps/conf/kafka/server.properties

2)關(guān)閉kafka

apps/svr/jdk/bin/jps -ml |grep 'kafka.Kafka' | awk '{print $1}' |xargs kill -9

3)驗(yàn)證kafka進(jìn)程

/apps/svr/jdk/bin/jps -ml |grep 'kafka.Kafka' | awk '{print $1}'

4)啟動(dòng)

/apps/svr/kafka/bin/kafka-server-start.sh -daemon /apps/conf/kafka/server.properties

/apps/svr/jdk/bin/jps -ml |grep 'kafka.Kafka' | awk '{print $1}'

5)觀察kafka數(shù)據(jù)是夠已經(jīng)完成了數(shù)據(jù)purge

(1)通過(guò)server.log查看purge過(guò)程

(2)直觀的觀察data目錄是夠存在2個(gè)小時(shí)之前的日志data

3、增加partitions

因?yàn)樵黾庸?jié)點(diǎn)，物理機(jī)器機(jī)器更多，需要增加partition的個(gè)數(shù)。

bin/kafka-topics.sh --zookeeper gd6-chenqun-zookeeper-003.idc.vip.com:2181/kafka --alter --topic all --partitions 192

4、重新分配parttion(reassign partitions)

1)獲取所有的topic

/apps/svr/kafka/bin/kafka-topics.sh --list --zookeeper gd6-chenuqn-zookeeper-001.idc.vip.com:2181/kafka

2) reassign partitions**

生成需要遷移的topic partitions信息，broker-list為所有的節(jié)點(diǎn)，包括新增節(jié)點(diǎn)。

./bin/kafka-reassign-partitions.sh --broker-list "1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24" --topics-to-move-json-file move.json --zookeeper gd6-chenqun-zookeeper-001.idc.vip.com:2181/kafka --generate

其中topics的json文件內(nèi)容為：

vim move.json

{"topics":

[{"topic": "log"},{"topic": "trace"},{"topic": "titan"}],

"version":1

}

3)使用上一步生成的建議partition json內(nèi)容進(jìn)行完成遷移

“Proposed partition reassignment configuration”后面的內(nèi)容保存到reassign.json文件中

….

bin/kafka-reassign-partitions.sh --broker-list "1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24" --reassignment-json-file reassign.json --zookeeper gd6-chenqun-zookeeper-001.idc.vip.com:2181/kafka --execute

4) 檢驗(yàn)partition的遷移狀態(tài)

bin/kafka-reassign-partitions.sh --verify --zookeeper gd6-chenqun-zookeeper-001.idc.vip.com:2181/logviewkafka

--reassignment-json-file reassign.json |grep -v successfully Status of partition reassignment: Reassignment of partition [all,43] is still in progress Reassignment of partition [all,11] is still in progress Reassignment of partition [all,107] is still in progress

5)修改參數(shù)，重啟kafka

sed -i 's/log.retention.hours=1/log.retention.hours=24/g' /apps/conf/kafka/server.properties

6)驗(yàn)證partition的分布

./bin/kafka-topics.sh --zookeeper gd6-chenqun-zookeeper-003.idc.vip.com:2181/logviewkafka --topic all --describe

總結(jié)

以上是生活随笔為你收集整理的kafka 重新分配节点_kafka扩容节点和partitions迁移的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。