kafka如何扩容分区Partition、并重新分区

小半 • 2023年1月15日下午2:27 • 技术随笔 • 阅读 170

1.扩容
在新的物理机上安装kafka程序，修改config/server.properties文件里的broker.id必须在集群中唯一，修改其他必要的配置项，其中zookeeper.connect配置项，写上kafka集群现在使用的zookeeper集群的地址。
然后启动kafka就可以加入到集群中了。
但是新加入的机器只能对新产生的topic起作用，对已有的topic在没有做处理前，是不会承担任何任务的，所以不会分担集群的压力。

2.重新分区Partition
假设有一个名为test的topic，只有1个partition，现在由于存储空间不足，需要重新分区。

3.修改topic的partitions

./bin/kafka-topics.sh --zookeeper 10.0.210.152:2181 --alter --topic test --partitions 6

现在topic有6个partition，但是数据还没有迁移过去

4.迁移数据
使用kafka提供的工具kafka-reassign-partitions.sh来迁移数据。迁移数据需要分三步做

第一步：生成迁移计划
先手动生成一个topic.json，内容如下。这里topic可以是一个列表，

{
"topics": [
{"topic": "test"}
],
"version": 1
}

执行如下语句，

./bin/kafka-reassign-partitions.sh --zookeeper 10.0.210.152:2181 --topics-to-move-json-file topic.json --broker-list "0,1,2,3,4" --generate

将topic.json里的topic迁移到broker-list列表里列的broker上，会得到一个执行计划

Current partition replica assignment
{"version":1,
"partitions":[....]
}
Proposed partition reassignment configuration
{"version":1,
"partitions":[.....]
}

新建一个文件reassignment.json，保存上边这些信息。其中Current partition replica assignment指当前的分区情况，Proposed partition reassignment configuration是计划的分区情况

第二步：迁移
执行如下命令

./bin/kafka-reassign-partitions.sh --zookeeper 10.0.210.152:2181 --reassignment-json-file reassignment.json --execute

第三步：验证

./bin/kafka-reassign-partitions.sh --zookeeper 10.0.210.152:2181 --reassignment-json-file reassignment.json --verify

第四步：随机查看topics是否迁移

./bin/kafka-topics.sh --zookeeper 10.0.210.152:2181 --describe --topic test

5.其他kafka常用命令

./bin/kafka-console-producer.sh --broker-list 10.0.210.152:9092 --topic test
./bin/kafka-console-consumer.sh --zookeeper 10.0.210.152:2181 --topic test --from-beginning
./bin/kafka-topics.sh --zookeeper 10.0.210.152:2181 --list
./bin/kafka-topics.sh --zookeeper 10.0.210.152:2181 --create --replication-factor 2 --partition 6 --topic test
./bin/kafka-topics.sh --zookeeper 10.0.210.152:2181 --delete --topic test
./bin/kafka-topics.sh --zookeeper 10.0.210.152:2181 --describe --topic test