kafka topic不均衡 (清理kafka topic消息)

以下是一个Python脚本，用于修复Kafka Topic分区分配不均匀的问题。该脚本会自动计算分区的分配方案，并将分配方案更新到Kafka集群中，以实现负载均衡。

```python

from kafka import KafkaAdminClient, KafkaConsumer

from kafka.admin import NewPartitionAssignment, NewPartitions, ConfigResource, ConfigResourceType

from typing import List, Dict

# Kafka集群中的Broker地址

bootstrap_servers = ['localhost:9092']

# Kafka Topic名称

topic_name = 'your_topic_name'

# 消费者组ID

group_id = 'your_consumer_group_id'

# 获取Topic的分区分配情况

def get_topic_partitions(topic_name: str) -> Dict[int, List[int]]:

admin_client = KafkaAdminClient(bootstrap_servers=bootstrap_servers)

topic_metadata = admin_client.describe_topics([topic_name])

topic_partitions = topic_metadata[0].partitions

partition_assignment = {}

for partition in topic_partitions:

replicas = partition.replicas

partition_assignment[partition.partition] = [replica.id for replica in replicas]

return partition_assignment

# 获取消费者组的消费情况

def get_consumer_offsets(topic_name: str, group_id: str) -> Dict[int, int]:

consumer = KafkaConsumer(topic_name, group_id=group_id, bootstrap_servers=bootstrap_servers)

partitions = consumer.partitions_for_topic(topic_name)

offsets = {}

for partition in partitions:

tp = (topic_name, partition)

offset = consumer.position(tp)

offsets[partition] = offset

return offsets

# 计算新的分区分配方案

def calculate_partition_assignment(partition_assignment: Dict[int, List[int]], offsets: Dict[int, int]) -> NewPartitionAssignment:

num_partitions = len(partition_assignment)

num_consumers = len(offsets)

avg = sum(offsets.values()) // num_consumers

assignments = [[] for i in range(num_consumers)]

for partition in sorted(offsets, key=offsets.get):

min_index = 0

min_val = sum([offsets[part] for part in assignments[0]])

for i in range(1, num_consumers):

val = sum([offsets[part] for part in assignments[i]])

if val < min_val:

min_index = i

min_val = val

assignments[min_index].append(partition)

new_assignment = {}

for i, consumer in enumerate(assignments):

for partition in consumer:

new_assignment[partition] = partition_assignment[partition] + [i]

return NewPartitionAssignment(new_assignment)

# 增加分区

def add_topic_partitions(topic_name: str, num_partitions: int):

admin_client = KafkaAdminClient(bootstrap_servers=bootstrap_servers)

topic_partitions = NewPartitions(total_count=num_partitions)

admin_client.create_partitions({'topic': topic_name, 'new_partitions': topic_partitions})

# 更新Topic的配置参数

def update_topic_config(topic_name: str, config: Dict[str, str]):

admin_client = KafkaAdminClient(bootstrap_servers=bootstrap_servers)

config_resource = ConfigResource(ConfigResourceType.TOPIC, topic_name)

admin_client.alter_configs({config_resource: config})

# 执行分区分配修复操作

def rebalance_topic_partitions(topic_name: str, group_id: str):

partition_assignment = get_topic_partitions(topic_name)

offsets = get_consumer_offsets(topic_name, group_id)

new_assignment = calculate_partition_assignment(partition_assignment, offsets)

admin_client = KafkaAdminClient(bootstrap_servers=bootstrap_servers)

admin_client.alter_partition_assignments({topic_name: new_assignment})

print(f'Topic {topic_name} partition assignment has been updated.')

# 示例：增加2个分区

add_topic_partitions(topic_name, 2)

# 示例：更新Topic的配置参数

update_topic_config(topic_name, {'max.message.bytes': '1048576'})

# 示例：执行分区分配修复操作

rebalance_topic_partitions(topic_name, group_id)

```

请注意，该脚本仅为示例代码，您需要根据实际情况进行修改和调整，以确保脚本的正确性和可靠性。同时，建议在执行脚本之前备份您的Kafka数据，以避免不可预见的错误和数据丢失。