說明:

在擴(kuò)大kafka集群時，我們需要滿足兩個要求：

1. 將指定的topic轉(zhuǎn)移到集群中添加的node上。
2. 將指定的topicpartition轉(zhuǎn)移到新的node上。

1. 將topic轉(zhuǎn)移到新的node上

假設(shè)現(xiàn)在一個kafka集群運(yùn)行三個broker，broker.id依次為101,102,103，隨后由于業(yè)務(wù)數(shù)據(jù)的突然增加，需要增加三個broker，broker.id依次為104,105,106.目的是使push。-token-將topic轉(zhuǎn)移到新的node上。

1、migration腳本-push-token-topic.json文件的內(nèi)容如下：

1. {
3. "topics":
5. [
7. {
9. "topic":"push-token-topic"
11. }
13. ],
15. "version":1
17. }

2、執(zhí)行腳本如下所示：

1. root@localhost:$./bin/kafka-reassign-partitions.sh--192.168.zookeper.225:2183--topics-to-move-json-filemigration-push-token-topic.json--broker-list"104,105,106"

json腳本的生成分配partitions 恢復(fù)備份使用：

Current partition replica assignment

{"version":1,"partitions":[{"topic":"cluster-switch-topic","partition":10,"replicas":[8]},{"topic":"cluster-switch-topic","partition":5,"replicas":[4]}，{"topic":"cluster-switch-topic","partition":3,"replicas":[5]}，{"topic":"cluster-switch-topic","partition":4,"replicas":[5]},{"topic":"cluster-switch-topic","partition":9,"replicas":[5]}，{"topic":"cluster-switch-topic","partition":1,"replicas":[5]},{"topic":"cluster-switch-topic","partition":11,"replicas":[4]}，{"topic":"cluster-switch-topic","partition":7,"replicas":[5]},{"topic":"cluster-switch-topic","partition":2,"replicas":[4]}，{"topic":"cluster-switch-topic","partition":0,"replicas":[4]},{"topic":"cluster-switch-topic","partition":6,"replicas":[4]}，{"topic":"cluster-switch-topic","partition":8,"replicas":[4]}]}

重新分配parttions的json腳本如下：

migration-topic-cluster-switch-topic.json

{"version":1,"partitions":[{"topic":"cluster-switch-topic","partition":10,"replicas":[5]}，{"topic":"cluster-switch-topic","partition":5,"replicas":[4]}，{"topic":"cluster-switch-topic","partition":4,"replicas":[5]}，{"topic":"cluster-switch-topic","partition":3,"replicas":[4]},{"topic":"cluster-switch-topic","partition":9,"replicas":[4]}，{"topic":"cluster-switch-topic","partition":1,"replicas":[4]},{"topic":"cluster-switch-topic","partition":11,"replicas":[4]}，{"topic":"cluster-switch-topic","partition":7,"replicas":[4]},{"topic":"cluster-switch-topic","partition":2,"replicas":[5]}，{"topic":"cluster-switch-topic","partition":0,"replicas":[5]},{"topic":"cluster-switch-topic","partition":6,"replicas":[5]}，{"topic":"cluster-switch-topic","partition":8,"replicas":[5]}]}

3、執(zhí)行：

1. root@localhost:$bin/kafka-reassign-partitions.sh--192.168.zookeper.225:2183--reassignment-json-filemigration-topic-cluster-switch-topic.json--execute

一個json格式文件expand將在執(zhí)行后生成。-cluster-reassignment.json

4、查詢執(zhí)行狀態(tài)：

1. bin/kafka-reassign-partitions.sh--192.168.zookeper.225:2183

不需要partion就可以在正常執(zhí)行后回到當(dāng)前的數(shù)據(jù)遷移狀態(tài)，信息狀態(tài)與下面相似。

1. Reassignmentofpartition[push-token-topic,0]completedsuccessfully//成功移動
2. Reassignmentofpartition[push-token-topic,1]isinprogress///代表數(shù)據(jù)在移動過程中的數(shù)據(jù)。
3. Reassignmentofpartition[push-token-topic,2]isinprogress
4. Reassignmentofpartition[push-token-topic,1]completedsuccessfully
5. Reassignmentofpartition[push-token-topic,2]completedsuccessfully

這樣做不會影響原來集群中的topic業(yè)務(wù)。

2.修改topic(replicats-factor)副本數(shù)量

假設(shè)push是初始的-token-topic是一個副本，為了提高可用性，需要改為兩個副本。

replicass腳本-update-push-token-topic.json文件的內(nèi)容如下：

{

"partitions":

[

{

"topic": "log.mobile_nginx",

"partition": 0,

"replicas": [101,102,104]

},

{

"topic": "log.mobile_nginx",

"partition": 1,

"replicas": [102,103,106]

}

],

"version":1

}

2、執(zhí)行：

1. root@localhost:$./bin/kafka-reassign-partitions.sh--192.168.zookeper.225:2183

3、verify

1. bin/kafka-reassign-partitions.sh--192.168.zookeper.225:2181

如下：

Status of partition reassignment: Reassignment of partition [log.mobile_nginx,0] completed successfully Reassignment of partition [log.mobile_nginx,1] completed successfully

自定義分區(qū)和轉(zhuǎn)移

1、The first step is to hand craft the custom reassignment plan in a json file-

> cat custom-reassignment.json {"version":1,"partitions":[{"topic":"foo1""partition":0,"replicas":[5,6]},{"topic":"foo2""partition":1,"replicas":[2,3]}]}

> bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file custom-reassignment.json --execute Current partition replica assignment {"version":1, "partitions":[{"topic":"foo1""partition":0,"replicas":[1,2]}, {"topic":"foo2""partition":1,"replicas":[3,4]}] } Save this to use as the --reassignment-json-file option during rollback Successfully started reassignment of partitions {"version":1, "partitions":[{"topic":"foo1""partition":0,"replicas":[5,6]}, {"topic":"foo2""partition":1,"replicas":[2,3]}] }

3、The --verify option can be used with the tool to check the status of the partition reassignment. Note that the same expand-cluster-reassignment.json (used with the --execute option) should be used with the --verify option

bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file custom-reassignment.json --verify Status of partition reassignment: Reassignment of partition [foo1,0] completed successfully Reassignment of partition [foo2,1] completed successfully

4.分區(qū)擴(kuò)展用法topic

a.首先擴(kuò)大分區(qū)數(shù)量，腳本如下：

比如:push-token-最初的topic分區(qū)數(shù)為12個，目前已增至15個。

root@localhost:$ ./bin/kafka-topics.sh --zookeeper 192.168.2.225:2183 --alter --partitions 15 --topic push-token-topic

b.設(shè)置topic分區(qū)副本

root@localhost:$ ./bin/kafka-reassign-partitions.sh --zookeeper 192.168.2.225:2183

--reassignment-json-file partitions-extension-push-token-topic.json --execute

腳本partitions-extension-push-token-topic.json文件的內(nèi)容如下：

{

"partitions":

[

{

"topic": "push-token-topic",

"partition": 12,

"replicas": [101,102]

},

{

"topic": "push-token-topic",

"partition": 13,

"replicas": [103,104]

},

{

"topic": "push-token-topic",

"partition": 14,

"replicas": [105,106]

}

],

"version":1

}