kafka的擴容難點：

1)主要在于增加機器之后，數據需要rebalance到新增的空閑節點，即把partitions遷移到空閑機器上。

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

2)kafka的集群的數據量加大，數據rebalance的時間較長。解決辦法是把log.retention.hours=1設置一小時(生產參數24小時)。

修改參數之后，重啟kakfa節點，kafka會主動purge 1小時之前的log數據。

以下是kafka_0.8.1.1版本kafkka集群擴容操作記錄，從16臺物理機擴容到24臺物理，partition數量由128個增加到192個：

kafka支持參數的動態修改，下面單獨設置某個topic的retention時間。

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

1、準備工作：

1) 驗證kafka節點是否正常加入集群。

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

2)檢查新增機器的zk信息中主機名是否準確，每個broker都要檢查

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

2、purge數據，使數據遷移更加快速

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

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

2)關閉kafka

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

3)驗證kafka進程

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

4)啟動

/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數據是夠已經完成了數據purge

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

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

3、增加partitions

因為增加節點，物理機器機器更多，需要增加partition的個數。

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為所有的節點，包括新增節點。

./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文件內容為：

vim move.json

{"topics":

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

"version":1

}

3)使用上一步生成的建議partition json內容進行完成遷移

“Proposed partition reassignment configuration”后面的內容保存到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) 檢驗partition的遷移狀態

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)修改參數，重啟kafka

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

6)驗證partition的分布

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

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。