kafka如何调整分区数量

5月 2nd, 2025

Kafka的分区数量可以通过以下几种方式进行：

1. 创建Topic时指定分区数量

在创建新的Topic时，可以直接指定分区数量。使用kafka-topics.sh脚本可以这样操作：

bin/kafka-topics.sh --create --topic your_topic_name --partitions 10 --bootstrap-server your_broker:9092 --replication-factor 3

--topic：指定Topic名称。
--partitions：指定分区数量。
--bootstrap-server：指定Kafka集群的Broker地址。
--replication-factor：指定副本因子。

2. 修改现有Topic的分区数量

对于已经存在的Topic，不能直接修改分区数量。需要通过以下步骤来实现：

a. 创建一个新的Topic

首先，创建一个具有所需分区数量的新Topic：

bin/kafka-topics.sh --create --topic new_topic_name --partitions desired_partitions --bootstrap-server your_broker:9092 --replication-factor 3

b. 复制数据

使用kafka-reassign-partitions.sh脚本将旧Topic的数据复制到新Topic中。首先，生成一个分区重分配计划：

bin/kafka-reassign-partitions.sh --topics-to-move-json-file old_topic_name.json --broker-list "broker1:9092,broker2:9092,broker3:9092" --generate --zookeeper zookeeper_host:2181

--topics-to-move-json-file：指定包含旧Topic信息的JSON文件。
--broker-list：指定Kafka集群的Broker地址列表。
--generate：生成分区重分配计划。
--zookeeper：指定Zookeeper地址。

然后，应用这个计划：

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

c. 验证数据

确保数据已经成功复制到新Topic中。

d. 删除旧Topic

最后，删除旧Topic：

bin/kafka-topics.sh --delete --topic old_topic_name --bootstrap-server your_broker:9092

3. 通过Kafka Admin API调整分区数量

Kafka Admin API也可以用来动态调整分区数量，但这种方式通常用于编程环境：

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewPartitions;
import org.apache.kafka.clients.admin.TopicDescription;

import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class KafkaPartitionResizer {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "your_broker:9092");

        try (AdminClient adminClient = AdminClient.create(props)) {
            NewPartitions newPartitions = NewPartitions.increaseTo(10);
            TopicDescription topicDescription = adminClient.describeTopics(Collections.singletonList("your_topic_name")).all().get().get("your_topic_name");
            topicDescription.partitions().forEach(partitionInfo -> {
                if (partitionInfo.partition() == 0) { // 假设只有一个分区
                    partitionInfo.partitions().set(partitionInfo.partitions().size(), newPartitions);
                }
            });

            adminClient.alterTopics(Collections.singletonList(topicDescription)).all().get();
        }
    }
}