kafka partition 扩容

站长

2024年06月09日 18:50 · 阅读数 30

在 Apache Kafka 中，扩展主题的分区数（Partition）是一个常见的任务。这需要注意的是，Kafka 并不会自动重新平衡现有的分区数据，因此在增加分区后，新分区将是空的，旧的分区数据不会自动迁移到新的分区。

下面是一个 Kafka 主题扩容的真实案例以及扩容脚本的设计。

案例描述

假设我们有一个 Kafka 主题 my-topic，当前有 3 个分区。由于业务增长，我们需要将分区数增加到 6 个。

步骤一：验证当前分区数

首先，验证主题的当前配置，以确认当前有 3 个分区。

kafka-topics.sh --describe --topic my-topic --bootstrap-server localhost:9092

输出示例：

Topic: my-topic	PartitionCount:3	ReplicationFactor:1	Configs:
	Topic: my-topic	Partition: 0	Leader: 1	Replicas: 1	Isr: 1
	Topic: my-topic	Partition: 1	Leader: 1	Replicas: 1	Isr: 1
	Topic: my-topic	Partition: 2	Leader: 1	Replicas: 1	Isr: 1

步骤二：增加分区数

使用 kafka-topics.sh 工具增加主题的分区数。

kafka-topics.sh --alter --topic my-topic --partitions 6 --bootstrap-server localhost:9092

步骤三：验证分区数增加

再次描述主题以验证分区数是否已增加到 6 个。

kafka-topics.sh --describe --topic my-topic --bootstrap-server localhost:9092

输出示例：

Topic: my-topic	PartitionCount:6	ReplicationFactor:1	Configs:
	Topic: my-topic	Partition: 0	Leader: 1	Replicas: 1	Isr: 1
	Topic: my-topic	Partition: 1	Leader: 1	Replicas: 1	Isr: 1
	Topic: my-topic	Partition: 2	Leader: 1	Replicas: 1	Isr: 1
	Topic: my-topic	Partition: 3	Leader: 1	Replicas: 1	Isr: 1
	Topic: my-topic	Partition: 4	Leader: 1	Replicas: 1	Isr: 1
	Topic: my-topic	Partition: 5	Leader: 1	Replicas: 1	Isr: 1

扩容脚本

以下是一个用于扩容 Kafka 主题分区数的 Bash 脚本示例：

#!/bin/bash

# Kafka broker address
BROKER="localhost:9092"

# Topic to be expanded
TOPIC="my-topic"

# New number of partitions
NEW_PARTITIONS=6

# Check current partition count
CURRENT_PARTITIONS=$(kafka-topics.sh --describe --topic $TOPIC --bootstrap-server $BROKER | grep "PartitionCount" | awk -F ':' '{print $2}' | tr -d '[:space:]')

echo "Current partition count for topic $TOPIC: $CURRENT_PARTITIONS"

if [ "$CURRENT_PARTITIONS" -lt "$NEW_PARTITIONS" ]; then
    echo "Increasing partitions for topic $TOPIC to $NEW_PARTITIONS"
    kafka-topics.sh --alter --topic $TOPIC --partitions $NEW_PARTITIONS
   
    if [ $? -eq 0 ]; then
        echo "Successfully increased the number of partitions for topic $TOPIC to $NEW_PARTITIONS."
    else
        echo "Failed to increase the number of partitions for topic $TOPIC."
        exit 1
    fi
else
    echo "The current number of partitions ($CURRENT_PARTITIONS) is already equal to or greater than the requested number of partitions ($NEW_PARTITIONS)."
fi

# Verify the new partition count
NEW_PARTITION_COUNT=$(kafka-topics.sh --describe --topic $TOPIC --bootstrap-server $BROKER | grep "PartitionCount" | awk -F ':' '{print $2}' | tr -d '[:space:]')

if [ "$NEW_PARTITION_COUNT" -eq "$NEW_PARTITIONS" ]; then
    echo "Verification successful: The topic $TOPIC now has $NEW_PARTITION_COUNT partitions."
else
    echo "Verification failed: The topic $TOPIC has $NEW_PARTITION_COUNT partitions instead of $NEW_PARTITIONS."
    exit 1
fi

echo "Partition expansion script completed successfully."

使用步骤

保存脚本： 将上述脚本内容保存到一个文件中，例如 expand_kafka_partitions.sh。
赋予执行权限： 确保脚本具有执行权限。
```
chmod +x expand_kafka_partitions.sh
```
运行脚本： 执行脚本来扩展 Kafka 主题的分区数。
```
./expand_kafka_partitions.sh
```

注意事项

无数据重新平衡： 扩展分区后，新创建的分区是空的，现有的数据不会自动重新平衡到新分区上。如果需要重新平衡数据，可能需要额外的步骤和工具，如 Kafka Streams 或者自定义的消费逻辑。
消费者处理： 确保消费者能够正确处理分区的变化。Kafka 消费者客户端通常会自动适应分区的增加，但在某些情况下，可能需要重新启动消费者应用以检测新的分区。

通过这些步骤和脚本，您可以有效地扩展 Kafka 主题的分区数，以应对不断增长的业务需求。

如何做到轻量级扩容？

轻量级扩容 Kafka 分区主要指尽量减少对现有系统的影响和操作复杂度。以下是一些具体步骤和脚本，以实现轻量级扩容：

轻量级扩容步骤

验证当前分区数： 了解当前分区数和主题的状态。
增加分区数： 使用 Kafka 提供的工具增加分区。
验证新分区数： 确保分区增加成功。
调整消费者配置（如果需要）： 确保消费者可以正确消费新分区的数据。

轻量级扩容脚本

以下是一个 Bash 脚本示例，用于轻量级扩容 Kafka 主题的分区数：

#!/bin/bash

# Kafka broker address
BROKER="localhost:9092"

# Topic to be expanded
TOPIC="my-topic"

# New number of partitions
NEW_PARTITIONS=6

# Function to check current partition count
check_partition_count() {
    kafka-topics.sh --describe --topic $TOPIC --bootstrap-server $BROKER | grep "PartitionCount" | awk -F ':' '{print $2}' | tr -d '[:space:]'
}

# Get the current partition count
CURRENT_PARTITIONS=$(check_partition_count)

echo "Current partition count for topic $TOPIC: $CURRENT_PARTITIONS"

if [ "$CURRENT_PARTITIONS" -lt "$NEW_PARTITIONS" ]; then
    echo "Increasing partitions for topic $TOPIC to $NEW_PARTITIONS"
    kafka-topics.sh --alter --topic $TOPIC --partitions $NEW_PARTITIONS --bootstrap-server $BROKER
    if [ $? -eq 0 ]; then
        echo "Successfully increased the number of partitions for topic $TOPIC to $NEW_PARTITIONS."
    else
        echo "Failed to increase the number of partitions for topic $TOPIC."
        exit 1
    fi
else
    echo "The current number of partitions ($CURRENT_PARTITIONS) is already equal to or greater than the requested number of partitions ($NEW_PARTITIONS)."
fi

# Verify the new partition count
NEW_PARTITION_COUNT=$(check_partition_count)

if [ "$NEW_PARTITION_COUNT" -eq "$NEW_PARTITIONS" ]; then
    echo "Verification successful: The topic $TOPIC now has $NEW_PARTITION_COUNT partitions."
else
    echo "Verification failed: The topic $TOPIC has $NEW_PARTITION_COUNT partitions instead of $NEW_PARTITIONS."
    exit 1
fi

echo "Partition expansion script completed successfully."

使用步骤

保存脚本： 将上述脚本内容保存到一个文件中，例如 lightweight_expand_kafka_partitions.sh。
赋予执行权限： 确保脚本具有执行权限。
```
chmod +x lightweight_expand_kafka_partitions.sh
```
运行脚本： 执行脚本来扩展 Kafka 主题的分区数。
```
./lightweight_expand_kafka_partitions.sh
```

注意事项

最小化停机时间： 增加分区的操作通常是即时的，不需要停机，但建议在低流量时段进行操作。
消费者重新平衡： Kafka 消费者会自动检测到新的分区并重新平衡。确保您的消费者是自动管理分区的，例如使用 Kafka 消费者组（Consumer Group）。
监控和验证： 扩容后，通过监控工具验证 Kafka 集群的健康状态和性能，确保没有出现异常。

通过上述步骤和脚本，您可以在不影响现有系统稳定性的情况下轻量级地扩容 Kafka 主题的分区数。

需要修改的地方

在轻量级扩容 Kafka 分区的过程中，一般情况下不需要修改 Kafka 的配置文件，因为 Kafka 提供的工具（如 kafka-topics.sh）可以直接通过命令行参数来完成分区的扩容。

不过，为了确保操作顺利进行并在生产环境中执行时更加安全和高效，您可能需要关注和验证以下几个方面的配置：

1. Kafka Broker 配置文件

通常情况下，Kafka Broker 的配置文件位于 server.properties 中。以下是一些可能需要检查的配置项：

zookeeper.connect: 确保连接到正确的 Zookeeper 集群。
listeners: 确保配置了正确的监听地址。
log.dirs: 确保有足够的存储空间来容纳新增分区的数据。

2. 主题级别配置

在扩容分区之前，可能需要检查并了解主题的当前配置，包括但不限于：

replication.factor: 确保副本因子合适。
min.insync.replicas: 确保在分区扩展后，仍然有足够的同步副本来维持高可用性。

3. 消费者和生产者配置

扩容分区后，消费者和生产者通常不需要重新配置，但需要确保它们能够正确处理新的分区。

消费者配置

group.id: 确保所有消费者在同一个消费组中，以便能够自动重新平衡。
auto.offset.reset: 在新分区上没有偏移时如何处理，通常设置为 earliest 或 latest。

生产者配置

partitioner.class: 确保生产者在新分区数目下能够正确分配消息。如果使用自定义分区器，可能需要验证其逻辑是否正确。

验证配置

在执行扩容操作之前和之后，可以通过以下命令验证配置：

kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name my-topic --describe

示例配置检查脚本

下面是一个示例脚本，用于检查 Kafka 配置文件中的重要参数：

#!/bin/bash

# Kafka config file path
KAFKA_CONFIG="/path/to/your/kafka/config/server.properties"

# Check important Kafka configuration parameters
echo "Checking Kafka configuration parameters..."

grep -E "zookeeper.connect|listeners|log.dirs" $KAFKA_CONFIG

if [ $? -eq 0 ]; then
    echo "Important Kafka configuration parameters are correctly set."
else
    echo "Please check your Kafka configuration file ($KAFKA_CONFIG) and ensure important parameters are set correctly."
    exit 1
fi

echo "Kafka configuration check completed."

执行脚本

保存上述脚本为 check_kafka_config.sh，赋予执行权限并运行：

chmod +x check_kafka_config.sh
./check_kafka_config.sh

通过这些步骤，您可以确保 Kafka 集群的配置正确并能够支持分区扩容操作，同时保证生产者和消费者能够正确处理扩容后的分区。

转载自:https://juejin.cn/post/7374419258397163532