kafka节点的服役和退役

假设集群只有hadoop102、hadoop103、hadoop104中有kafka的节点，现在需要在hadoop105中新增一个节点，并且需要将部分主题的数据迁移到新增的节点中。或者将hadoop105节点退役掉，需要将105中的主题的数据迁移到现役的broker中。

服役新节点

1. 在hadoop105上安装kafka，或者通过rsync等命令传输过去（需要注意的是通过命令传输过去的kafka需要将data、logs等删除干净）；
2. 启动hadoop102、hadoop103、hadoop104上的kafka集群。
3. 修改haodoop105中kafka的broker.id为3
4. 单独启动hadoop105中的kafka

 [wyp@hadoop105 kafka]$ bin/kafka-server-start.sh -daemon ./config/server.properties

5. 创建一个要均衡的主题

[wyp@hadoop102 kafka]$ vim topics-to-move.json 

{
    "topics": [

        {"topic": "first"}

    ],

    "version": 1
}

5. 生成一个负载均衡的计划

[wyp@hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092  --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2,3" --generate

Current partition replica assignment
{"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[0,2,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[2,1,0],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[1,0,2],"log_dirs":["any","any","any"]}]}

Proposed partition reassignment configuration
{"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2,3,0],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[3,0,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[0,1,2],"log_dirs":["any","any","any"]}]}

6. 创建副本存储计划（所有副本存储在broker0、broker1、broker2、broker3中）。

[wyp@hadoop102 kafka]$ vim increase-replication-factor.json

将上方系统生成的Proposed partition reassignment configuration下的内容复制到 increase-replication-factor.json

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

7. 执行副本存储计划

[wyp@hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute

8. 验证副本存储计划(可以通过命令的形式查看或者去105节点上查看)

[wyp@hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --verify

执行后看到如下输出就是完成节点的数据均衡了。

Status of partition reassignment:
Reassignment of partition first-0 is complete.
Reassignment of partition first-1 is complete.
Reassignment of partition first-2 is complete.

Clearing broker-level throttles on brokers 0,1,2,3
Clearing topic-level throttles on topic first

退役旧节点

1. 创建一个要均衡的主题。

[wyp@hadoop102 kafka]$ vim topics-to-move.json 

{
    "topics": [

        {"topic": "first"}

    ],

    "version": 1
}

2. 创建执行计划

[wyp@hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092  --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2" --generate

Current partition replica assignment
{"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2,0,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[3,1,2],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[0,2,3],"log_dirs":["any","any","any"]}]}

Proposed partition reassignment configuration
{"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2,0,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[1,2,0],"log_dirs":["any","any","any"]}]}

3. 创建副本存储计划

[wyp@hadoop102 kafka]$ vim increase-replication-factor.json
# 把上方的Proposed partition reassignment configuration复制到文件中
{"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2,0,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[1,2,0],"log_dirs":["any","any","any"]}]}

4. 执行副本存储计划

[wyp@hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute

5. 验证副本存储计划

[wyp@hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --verify

Status of partition reassignment:
Reassignment of partition first-0 is complete.
Reassignment of partition first-1 is complete.
Reassignment of partition first-2 is complete.

Clearing broker-level throttles on brokers 0,1,2,3
Clearing topic-level throttles on topic first

总结：


kafka节点的服役和退役的节点操作是相同，均是将主题中的数据均匀的分配到新增或者删除节点之外的borker中。

操作步骤：

(1)创建需要均衡的主题；

(2)生成负载均衡的计划；

(3)创建并执行生成的负载均衡的计划；

(4)验证计划的执行情况。

需要注意的是，服役节点时，新增的节点的brokerId需要修改，rsync复制过去的需要将data、logs中的内容删除掉。