怎样用脚本调整Kafka分区？

wen 实用脚本 2026-07-21 4

高效管理Kafka分区：脚本化调整的完整实战指南

目录导读

为什么需要脚本调整Kafka分区？
核心概念：分区、副本与脚本操作基础
实战脚本：增加分区（含代码与注意事项）
实战脚本：重新分配分区与副本（自动化迁移）
常见问题与问答（Q&A）
监控与验证：脚本执行后的健康检查

为什么需要脚本调整Kafka分区？

在Kafka生产环境中，分区数是影响吞吐量与数据分布的关键参数，随着业务增长,可能出现以下场景：

怎样用脚本调整Kafka分区？

单分区瓶颈：某个分区写入速度远超其他分区，导致数据倾斜。
节点扩容：新增Broker后，需要将旧节点上的分区副本均匀分配到新节点。
运维自动化：手动调整多个Topic分区效率低下,且容易出错。

答案： 通过脚本化调整，可以实现批量分区修改、动态副本重分配、以及自动化运维,避免人工操作带来的风险。

核心概念：分区、副本与脚本操作基础

1 分区与副本的关系

分区（Partition）：每个Topic可被拆分为多个分区，每个分区是一个有序消息队列。
副本（Replica）：分区具备多个副本（默认1个Leader + N个Follower），用于高可用。
分区数：决定消息并发写入能力,但过多分区会消耗更多文件句柄。

2 脚本工具介绍

Kafka官方提供两类核心脚本（均在bin/目录下）：

kafka-topics.sh：用于创建、删除、修改Topic分区数。
kafka-reassign-partitions.sh：用于生成分区迁移计划并执行副本重分配。

注意：分区数不能减少，只能增加（除非删除Topic重建）。

实战脚本：增加分区（含代码与注意事项）

1 基本命令结构

# 语法：增加某个Topic的分区数至N
bin/kafka-topics.sh --bootstrap-server localhost:9092 \
  --alter --topic my-topic --partitions <目标分区数>

2 真实案例：将Topic `orders` 从3个分区增加到6个

./kafka-topics.sh --bootstrap-server broker1:9092,broker2:9092 \
  --alter --topic orders --partitions 6

3 脚本自动化：批量处理多个Topic

编写Shell脚本batch_add_partitions.sh：

#!/bin/bash
BOOTSTRAP_SERVERS="broker1:9092,broker2:9092"
TOPICS_FILE="topics_partitions.txt"  # 格式: topic_name new_partition_count
while IFS= read -r line; do
    topic=$(echo "$line" | awk '{print $1}')
    new_part=$(echo "$line" | awk '{print $2}')
    echo "处理 Topic: $topic -> 分区数: $new_part"
    ./kafka-topics.sh --bootstrap-server $BOOTSTRAP_SERVERS \
      --alter --topic "$topic" --partitions "$new_part"
    if [ $? -eq 0 ]; then
        echo "[成功] $topic 调整完毕"
    else
        echo "[失败] $topic 调整失败，请检查日志"
    fi
done < "$TOPICS_FILE"

4 注意事项

数据一致性：增加分区后，原有消息仍保留在旧分区，新消息将按新分区策略写入。
Key-based分区：如果生产者使用Key分区（如hash(key) % 分区数），增加分区会导致同一Key的消息分散到不同分区，破坏顺序性。
消费者组：增加分区后，消费者组会自动触发rebalance,需确保消费端逻辑兼容。

实战脚本：重新分配分区与副本（自动化迁移）

1 场景需求

当新Broker加入集群时,需要将部分分区副本从旧节点迁移到新节点以平衡负载。

2 脚本步骤（三阶段）

阶段1：生成迁移计划（JSON文件）

# 生成一个计划，将所有分区均匀分布到所有节点
./kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
  --generate --broker-list 1,2,3 --topics-to-move-json-file topics.json

其中topics.json内容示例：

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

脚本输出一个proposed-assignment.json文件。

阶段2：执行迁移

./kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
  --execute --reassignment-json-file proposed-assignment.json

阶段3：验证迁移状态

./kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
  --verify --reassignment-json-file proposed-assignment.json

3 全自动脚本（含进度监控）

#!/bin/bash
# auto_rebalance.sh - 自动将负载均匀分布到所有Broker
BOOTSTRAP="broker1:9092,broker2:9092"
TOPIC_LIST=$1   #  "orders,payments,inventory"
if [ -z "$TOPIC_LIST" ]; then
    echo "用法: $0 \"topic1,topic2\""
    exit 1
fi
echo '{"version":1,"topics":[' > topics.json
first=true
for topic in $(echo $TOPIC_LIST | tr ',' ' '); do
    if [ "$first" = true ]; then
        first=false
    else
        echo ',' >> topics.json
    fi
    echo '{"topic":"'$topic'"}' >> topics.json
done
echo ']}' >> topics.json
# 获取所有Broker ID
BROKER_IDS=$(./zookeeper-shell.sh localhost:2181 <<< "ls /brokers/ids" | tail -n1 | tr -d '[] ')
BROKER_LIST=$(echo $BROKER_IDS | tr ',' ' ' | paste -sd ',')
echo "生成迁移计划，目标Broker: $BROKER_LIST"
./kafka-reassign-partitions.sh --bootstrap-server $BOOTSTRAP \
  --generate --broker-list $BROKER_LIST --topics-to-move-json-file topics.json \
  | grep -A 100 '"proposed-assignment.json"' > plan.json
echo "执行迁移..."
./kafka-reassign-partitions.sh --bootstrap-server $BOOTSTRAP \
  --execute --reassignment-json-file plan.json
# 循环检查直到完成
while true; do
    status=$(./kafka-reassign-partitions.sh --bootstrap-server $BOOTSTRAP \
      --verify --reassignment-json-file plan.json | grep "has finished" | wc -l)
    if [ "$status" -gt 0 ]; then
        echo "所有分区迁移完成！"
        break
    fi
    sleep 5
    echo "迁移进行中..."
done

常见问题与问答（Q&A）

Q1：增加分区后，已有消费组会丢失数据吗？

A：不会。 增加分区只影响后续消息的路由，旧分区中的消息仍按原偏移量消费,消费组需重新平衡后才会消费新分区。

Q2：分区数可以无限增加吗？

A：理论上可以，但受限于：

每个分区对应一个日志段文件和副本同步线程，过多分区会消耗大量文件句柄和内存。
Producer与Consumer的连接数随分区数线性增长,建议单个Broker分区数不超过4000。

Q3：脚本迁移期间，生产者和消费者需要暂停吗？

A：不需要。 Kafka的副本迁移是“在线”的，Leader切换期间可能产生几毫秒的延迟，但不会丢失消息,生产者会自动重试失败的请求。

Q4：如何监控脚本执行是否正常？

A：

查看Broker日志：grep "reassignment" /var/log/kafka/server.log
使用JMX指标：kafka.controller:type=KafkaController,name=ReassignPartitionsMetric

Q5：如果迁移计划失败，如何回滚？

A：执行kafka-reassign-partitions.sh --execute时可以指定--throttle限制迁移速率，失败后只需删除中间状态文件（如plan.json），然后重新生成计划,Kafka在重启后会自动停止未完成的迁移任务。

监控与验证：脚本执行后的健康检查

1 检查分区分布

./kafka-topics.sh --bootstrap-server localhost:9092 \
  --describe --topic orders | grep "Replicas:"

输出类似：

Topic: orders  Partition: 0  Leader: 1  Replicas: 1,2  Isr: 1,2
Topic: orders  Partition: 1  Leader: 2  Replicas: 2,3  Isr: 2,3

2 验证每个Broker的负载

使用kafka-log-dirs.sh查看磁盘使用：

./kafka-log-dirs.sh --bootstrap-server localhost:9092 \
  --describe --topic-list orders

3 性能测试（可选）

使用kafka-producer-perf-test.sh模拟写入：

./kafka-producer-perf-test.sh --topic orders --num-records 100000 \
  --record-size 1000 --throughput -1 --producer-props bootstrap.servers=localhost:9092

脚本调整Kafka分区的核心在于 理解业务需求（是否需要保持Key顺序）、精确控制迁移负载（使用--throttle限制带宽）、以及 持续监控，建议先在测试环境用脚本验证流程，再投入生产，通过本文的Shell脚本模板,你可以快速构建符合自身集群的自动化运维流水线。