如何通过Python-Kafka Admin Client创建Kafka主题及相关Admin工具咨询

Q1：是否存在可在Python程序中创建/删除Kafka主题的Python Kafka Admin客户端？Confluent是否有对应的Python Admin API？ #

当然有啦！目前有两个主流的Python Kafka库都完整支持Admin API功能，完全能满足你创建、删除主题的需求：

• confluent-kafka-python：这是Confluent官方维护的Python Kafka客户端，从1.4.0版本开始正式支持Admin API，功能和Kafka原生AdminClient对齐度极高，稳定性也有保障。
• kafka-python：社区维护的纯Python Kafka库，同样提供了AdminClient类，支持主题的创建、删除、查询等全流程操作。

你之前可能接触的是比较旧的版本，现在这两个库的Admin功能都已经很完善了~

Q2：如何使用Python-Kafka Admin Client创建Kafka主题？ #

下面分别给出两个主流库的实战代码示例，你可以根据自己的技术栈选择：

方式1：使用Confluent官方客户端 #

首先确保安装最新版本的库：

pip install confluent-kafka --upgrade

创建主题的代码示例：

from confluent_kafka.admin import AdminClient, NewTopic

def create_kafka_topic(bootstrap_servers, topic_name, num_partitions=1, replication_factor=1):
    # 初始化AdminClient，配置Kafka集群地址
    admin_client = AdminClient({"bootstrap.servers": bootstrap_servers})
    
    # 定义要创建的主题，可指定分区数、副本数及自定义配置
    new_topic = NewTopic(
        topic=topic_name,
        num_partitions=num_partitions,
        replication_factor=replication_factor
        # 可选：添加主题配置，比如{"cleanup.policy": "compact"}
    )
    
    # 发送主题创建请求
    fs = admin_client.create_topics([new_topic])
    
    # 等待操作完成，处理结果
    for topic, f in fs.items():
        try:
            f.result()  # 阻塞等待主题创建完成
            print(f"主题 {topic} 创建成功！")
        except Exception as e:
            print(f"创建主题 {topic} 失败: {e}")

# 调用示例
if __name__ == "__main__":
    # 替换为你的Kafka集群地址
    create_kafka_topic("localhost:9092", "test_user_topic", num_partitions=3, replication_factor=1)

方式2：使用kafka-python库 #

先安装库：

pip install kafka-python --upgrade

创建主题的代码示例：

from kafka.admin import KafkaAdminClient, NewTopic

def create_kafka_topic(bootstrap_servers, topic_name, num_partitions=1, replication_factor=1):
    # 初始化AdminClient
    admin_client = KafkaAdminClient(
        bootstrap_servers=bootstrap_servers,
        client_id='kafka_admin_demo'
    )
    
    # 构造主题列表
    topic_list = [
        NewTopic(
            name=topic_name,
            num_partitions=num_partitions,
            replication_factor=replication_factor
            # 可选：自定义主题配置，比如topic_configs={"retention.ms": "86400000"}
        )
    ]
    
    # 执行主题创建操作
    try:
        admin_client.create_topics(new_topics=topic_list, validate_only=False)
        print(f"主题 {topic_name} 创建成功！")
    except Exception as e:
        print(f"创建主题 {topic_name} 失败: {e}")

# 调用示例
if __name__ == "__main__":
    create_kafka_topic("localhost:9092", "test_order_topic", num_partitions=2, replication_factor=1)

注意事项 #

• 确保执行操作的客户端拥有Kafka集群的Create主题权限；
• 副本数replication_factor不能超过集群的Broker数量，否则会创建失败；
• 如果集群开启了ACL权限控制，需要提前配置好对应的权限规则。

内容的提问来源于stack exchange，提问作者Bharat