kafka集群的迁移与扩容(转)

前言

此篇文章转载自 Kafka学习(六)Kafka集群的迁移与扩容) ,目前很符合我的预期了,如有需要,可以直接到原站点查看。

Kafka的集群扩容实际上就是把Topic的Partition移动到新加的集群节点上。我们只需要copy一份Kafka的安装目录到新的节点机器上,修改一下相关配置文件(broker.id,logs.dir等等),但是新添加的Kafka节点机器是不会自动分配数据的,所以需要我们手动操作。

Kafka的扩容可以细分为以下三种:

  • 增加节点
  • 增加Topic副本
  • 增加分区

具体的步骤有两种方式:

  • 通过–topics-to-move-json-file和–broker-list批量生成新的Topic分区信息,然后根据该信息执行转移操作。
  • 手动写要移动的topic信息,更灵活,但是在大量Topic和Partition的情况下非常繁琐并且容易出错。
1
2
3
4
5
6
7
8
9
kafka-reassign-partitions.sh命令的三种模式

generate模式:给需要重新分配的Topic,自动生成reassign plan(执行计划),但并不执行
execute模式:根据指定的reassign plan(json文件)重新分配partition或者replication
verify模式:检查重新分配是否完成

$ kafka-reassign-partitions.sh --zookeeper localhost:2181 --topics-to-move-json-file json_file --broker-list "brokerIds" --generate
$ kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file json_file --execute
$ kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file json_file --verify

Kafka动态增加节点(Node)

新添加的Kafka节点机器是不会自动分配数据的,所以无法分担集群的负载,除非我们新建一个Topic,此时新的Topic会使用新添加的Kafka节点机器。如果我们想新添加的Kafka节点机器能够分担集群存储,需要手动将部分分区移动到新添加的Kafka节点机器上。

我们原来的Kafka节点分别是0,1,2,现在要加入3,4两个新节点

topicMove.json

1
2
3
4
5
6
{
"topics":[
{"topic":"logstash_test"}
],
"version":1
}

生成迁移的计划

1
2
3
4
5
6
7
$ kafka-reassign-partitions.sh --zookeeper localhost:2181 --topics-to-move-json-file topicMove.json --broker-list "0,1,2,3,4" --generate
Current partition replica assignment

{"version":1,"partitions":[{"topic":"logstash_test","partition":3,"replicas":[1,2,0]},{"topic":"logstash_test","partition":1,"replicas":[2,1,0]},{"topic":"logstash_test","partition":0,"replicas":[1,2,0]},{"topic":"logstash_test","partition":2,"replicas":[0,2,1]},{"topic":"logstash_test","partition":5,"replicas":[0,1,2]},{"topic":"logstash_test","partition":4,"replicas":[2,0,1]}]}
Proposed partition reassignment configuration

{"version":1,"partitions":[{"topic":"logstash_test","partition":3,"replicas":[2,0,1]},{"topic":"logstash_test","partition":1,"replicas":[0,3,4]},{"topic":"logstash_test","partition":0,"replicas":[4,2,3]},{"topic":"logstash_test","partition":2,"replicas":[1,4,0]},{"topic":"logstash_test","partition":5,"replicas":[4,3,0]},{"topic":"logstash_test","partition":4,"replicas":[3,1,2]}]}

上面是原来的所有partition在各个节点的分布情况,下面是加入4,5两个新节点之后所有partition在各个节点的分布情况

将生成的执行计划保存为add_node.json文件

重新分配partition(其实这里是添加新的节点)

1
2
3
4
5
6
7
$ kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file ~/Downloads/add_node.json --execute
Current partition replica assignment

{"version":1,"partitions":[{"topic":"logstash_test","partition":3,"replicas":[1,2,0]},{"topic":"logstash_test","partition":1,"replicas":[2,1,0]},{"topic":"logstash_test","partition":0,"replicas":[1,2,0]},{"topic":"logstash_test","partition":2,"replicas":[0,2,1]},{"topic":"logstash_test","partition":5,"replicas":[0,1,2]},{"topic":"logstash_test","partition":4,"replicas":[2,0,1]}]}

Save this to use as the --reassignment-json-file option during rollback
Successfully started reassignment of partitions {"version":1,"partitions":[{"topic":"logstash_test","partition":0,"replicas":[4,2,3]},{"topic":"logstash_test","partition":4,"replicas":[3,1,2]},{"topic":"logstash_test","partition":5,"replicas":[4,3,0]},{"topic":"logstash_test","partition":2,"replicas":[1,4,0]},{"topic":"logstash_test","partition":3,"replicas":[2,0,1]},{"topic":"logstash_test","partition":1,"replicas":[0,3,4]}]}

查看执行的状态

1
2
$ kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file ~/Downloads/add_node.json --verify
ERROR: Assigned replicas (X,X) don't match the list of replicas for reassignment (X,X) for partition [logstash_test,1]

假设出现类似这样的错误,他并不是真的出错,而是指目前仍在复制数据中。再过一段时间再运行verify命令,他就会消失(加入完成拷贝)

当然也可以不使用generate先生成执行计划,而是自己直接手动编辑生成add_node.json文件内容,然后直接execute执行,但是这样容易出错,Kafka节点比较少的时候推荐使用。

注意:其实通过generate的结果我们也可以看出,其实不管是扩容,减容,迁移,其实都是重新分配Topic或者Partition的过程。json文件的内容都是指定Topic下的Partition要移动到哪个Node上。

Kafka动态增加Topic副本(Replication)

查看当前node_log的Topic信息

1
2
3
$ kafka-topics.sh --describe --zookeeper localhost:2181 --topic node_log
Topic:node_log PartitionCount:1 ReplicationFactor:1 Configs:
Topic: node_log Partition: 0 Leader: 2 Replicas: 2 Isr: 2

add_replication.json

1
2
3
4
5
6
7
8
9
10
11
12
13
14
{
"version":1
"partitions":[
{
"topic":"node_log",
"partition":0,
"replicas":[
0,
1,
2
]
}
]
}

重新分配partition(其实这里是添加副本replication)

1
2
3
4
5
6
7
$ kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file ~/Downloads/add_replication.json --execute
Current partition replica assignment

{"version":1,"partitions":[{"topic":"node_log","partition":0,"replicas":[2]}]}

Save this to use as the --reassignment-json-file option during rollback
Successfully started reassignment of partitions {"version":1,"partitions":[{"topic":"node_log","partition":0,"replicas":[0,1,2]}]}

查看执行的状态

1
2
3
$ kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file ~/Downloads/add_replication.json --verify
Status of partition reassignment:
Reassignment of partition [node_log,0] completed successfully

如果遇到下面的错误很有可能是json文件格式有错误,仔细检查修正重新运行即可

1
2
3
4
5
Partitions reassignment failed due to Partition reassignment data file add_replication.json is empty
kafka.common.AdminCommandFailedException: Partition reassignment data file add_replication.json is empty
at kafka.admin.ReassignPartitionsCommand$.executeAssignment(ReassignPartitionsCommand.scala:120)
at kafka.admin.ReassignPartitionsCommand$.main(ReassignPartitionsCommand.scala:52)
at kafka.admin.ReassignPartitionsCommand.main(ReassignPartitionsCommand.scala)

查看添加replication后的node_log的Topic信息

1
2
3
$ kafka-topics.sh --describe --zookeeper localhost:2181 --topic node_log
Topic:node_log PartitionCount:1 ReplicationFactor:2 Configs:
Topic: node_log Partition: 0 Leader: 0 Replicas: 0,1,2 Isr: 1,0,2

Kafka动态增加分区(Parition)

查看当前node_log的Topic信息

1
2
3
$ kafka-topics.sh --describe --zookeeper localhost:2181 --topic node_log
Topic:node_log PartitionCount:1 ReplicationFactor:2 Configs:
Topic: node_log Partition: 0 Leader: 0 Replicas: 0,1,2 Isr: 1,0,2

add_partition.json

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
{
"version":1,
"partitions":[
{
"topic":"node_log",
"partition":0,
"replicas":[
0,
1,
2
]
},
{
"topic":"node_log",
"partition":1,
"replicas":[
0,
1,
2
]
},
{
"topic":"node_log",
"partition":2,
"replicas":[
0,
1,
2
]
}
]
}

重新分配partition

1
2
3
4
5
6
7
8
9
$ kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file ~/Downloads/add_partition.json --execute
Current partition replica assignment

{"version":1,"partitions":[{"topic":"node_log","partition":0,"replicas":[0,1,2]}]}

Save this to use as the --reassignment-json-file option during rollback
[2016-12-30 12:22:17,195] ERROR Skipping reassignment of partition [node_log,1] since it doesn't exist (kafka.admin.ReassignPartitionsCommand)
[2016-12-30 12:22:17,207] ERROR Skipping reassignment of partition [node_log,2] since it doesn't exist (kafka.admin.ReassignPartitionsCommand)
Successfully started reassignment of partitions {"version":1,"partitions":[{"topic":"node_log","partition":0,"replicas":[0,1,2]},{"topic":"node_log","partition":1,"replicas":[0,1,2]},{"topic":"node_log","partition":2,"replicas":[0,1,2]}]}

查看执行的状态

1
2
3
4
5
6
7
$ kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file ~/Downloads/add_partition.json --verify
Status of partition reassignment:
ERROR: Assigned replicas (1,2,0) don't match the list of replicas for reassignment (0,1,2) for partition [node_log,1]
ERROR: Assigned replicas (2,0,1) don't match the list of replicas for reassignment (0,1,2) for partition [node_log,2]
Reassignment of partition [node_log,0] completed successfully
Reassignment of partition [node_log,1] failed
Reassignment of partition [node_log,2] failed

报错是因为我们只有一个partition0,没有partition1,partition2,所以跳过了重新分配partition的过程。

我们需要先增加partition的数量,我们把partition的数量变成6个

1
2
3
$ kafka-topics.sh --zookeeper localhost:2181 --alter --topic node_log --partitions 6
WARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected
Adding partitions succeeded!

查看添加partition后的node_log的Topic信息

1
2
3
4
5
6
7
8
$ kafka-topics.sh --describe --zookeeper localhost:2181 --topic node_log
Topic:node_log PartitionCount:6 ReplicationFactor:3 Configs:
Topic: node_log Partition: 0 Leader: 0 Replicas: 0,1,2 Isr: 1,0,2
Topic: node_log Partition: 1 Leader: 1 Replicas: 1,2,0 Isr: 1,2,0
Topic: node_log Partition: 2 Leader: 2 Replicas: 2,0,1 Isr: 2,0,1
Topic: node_log Partition: 3 Leader: 0 Replicas: 0,2,1 Isr: 0,2,1
Topic: node_log Partition: 4 Leader: 1 Replicas: 1,0,2 Isr: 1,0,2
Topic: node_log Partition: 5 Leader: 2 Replicas: 2,1,0 Isr: 2,1,0

重新执行reassign,然后查看执行状态

1
2
3
4
5
$ kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file ~/Downloads/add_partition.json --verify
Status of partition reassignment:
Reassignment of partition [node_log,0] completed successfully
Reassignment of partition [node_log,1] completed successfully
Reassignment of partition [node_log,2] completed successfully

重新分配执行完成之后,再次查看node_log的Topic信息

1
2
3
4
5
6
7
8
9
$ kafka-topics.sh --describe --zookeeper localhost:2181 --topic node_log

Topic:node_log PartitionCount:6 ReplicationFactor:3 Configs:
Topic: node_log Partition: 0 Leader: 0 Replicas: 0,1,2 Isr: 1,2,0
Topic: node_log Partition: 1 Leader: 0 Replicas: 0,1,2 Isr: 1,2,0
Topic: node_log Partition: 2 Leader: 0 Replicas: 0,1,2 Isr: 1,2,0
Topic: node_log Partition: 3 Leader: 0 Replicas: 0,2,1 Isr: 1,2,0
Topic: node_log Partition: 4 Leader: 1 Replicas: 1,0,2 Isr: 1,2,0
Topic: node_log Partition: 5 Leader: 2 Replicas: 2,1,0 Isr: 1,2,0

执行reassign之后,我们发现0-3的partition的leader都是0,这是因为我们之前突然扩展partition到6个,而我们在reassign的之后只指定了0-2的partition的分配,我们修改一下add_partition.json文件如下,将6个partition均匀的分布在我们3个broker节点上,每个partition有2个replication。

add_partition.json

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
{
"version":1,
"partitions":[
{
"topic":"node_log",
"partition":0,
"replicas":[
0,
1
]
},
{
"topic":"node_log",
"partition":1,
"replicas":[
1,
2
]
},
{
"topic":"node_log",
"partition":2,
"replicas":[
2,
0
]
},
{
"topic":"node_log",
"partition":3,
"replicas":[
0,
1
]
},
{
"topic":"node_log",
"partition":4,
"replicas":[
1,
2
]
},
{
"topic":"node_log",
"partition":5,
"replicas":[
2,
0
]
}
]
}

再次查看reassign之后的Topic情况

1
2
3
4
5
6
7
8
$ kafka-topics.sh --describe --zookeeper localhost:2181 --topic node_log
Topic:node_log PartitionCount:6 ReplicationFactor:2 Configs:
Topic: node_log Partition: 0 Leader: 0 Replicas: 0,1 Isr: 0,1
Topic: node_log Partition: 1 Leader: 1 Replicas: 1,2 Isr: 1,2
Topic: node_log Partition: 2 Leader: 0 Replicas: 2,0 Isr: 0,2
Topic: node_log Partition: 3 Leader: 0 Replicas: 0,1 Isr: 0,1
Topic: node_log Partition: 4 Leader: 1 Replicas: 1,2 Isr: 1,2
Topic: node_log Partition: 5 Leader: 2 Replicas: 2,0 Isr: 0,2

我们发现还是有点小问题,partition2的leader仍然是0。这里先普及一下assigned replicas和preferred replica

assigned replicas和preferred replica

每个partitiion的所有replicas叫做”assigned replicas”,”assigned replicas”中的第一个replicas叫”preferred replica”,刚创建的topic一般”preferred replica”是leader。leader replica负责所有的读写。但随着时间推移,broker可能会停机,会导致leader迁移,导致机群的负载不均衡。

我们这里preferred replica已经是2了,但是leader却不是2,这样我们需要重新选举一下leader,需要使用kafka-preferred-replica-election.sh来调整。

两种操作方式:

  • 对所有Topics进行操作
1
$ kafka-preferred-replica-election.sh --zookeeper localhost:2181
  • 对某个Topic进行操作(json文件中指定要操作的Topic)
1
$ kafka-preferred-replica-election.sh --zookeeper localhost:2181 --path-to-json-file json_file

注意:这里我们只需要调整node_log的Topic的partition2的leader。

leaderTopic.json

1
2
3
4
5
6
{
"partitions":
[
{"topic":"node_log","partition":2}
]
}

重新选举leader

1
2
$ kafka-preferred-replica-election.sh --zookeeper localhost:2181 --path-to-json-file leaderTopic.json
Successfully started preferred replica election for partitions Set([node_log,2])

再次查看Topic情况,发现leader也是我们所期望的均匀分配了

1
2
3
4
5
6
7
8
$ kafka-topics.sh --describe --zookeeper localhost:2181 --topic node_log
Topic:node_log PartitionCount:6 ReplicationFactor:2 Configs:
Topic: node_log Partition: 0 Leader: 0 Replicas: 0,1 Isr: 0,1
Topic: node_log Partition: 1 Leader: 1 Replicas: 1,2 Isr: 2,1
Topic: node_log Partition: 2 Leader: 2 Replicas: 2,0 Isr: 2,0
Topic: node_log Partition: 3 Leader: 0 Replicas: 0,1 Isr: 0,1
Topic: node_log Partition: 4 Leader: 1 Replicas: 1,2 Isr: 2,1
Topic: node_log Partition: 5 Leader: 2 Replicas: 2,0 Isr: 0,2
0%