如何使用 Python-kafka 管理客户端在 Kafka 中创建主题？答案

【问题标题】：How to create topic in Kafka with Python-kafka admin client?如何使用 Python-kafka 管理客户端在 Kafka 中创建主题？
【发布时间】：2018-10-11 02:43:10
【问题描述】：

是否有任何 Python kafka 管理客户端可用于从 python 程序创建主题/删除主题？我找到了一些 python api，但它们都没有可用的 Admin api？

confluent 有 python admin api 吗？

【问题讨论】：

标签： apache-kafka kafka-consumer-api kafka-producer-api confluent-platform

【解决方案1】：

Confluent Python Kafka Client 确实有管理员支持。

查看此example 了解如何使用它。

【讨论】：

【解决方案2】：

from kafka.admin import KafkaAdminClient, NewTopic
admin_client = KafkaAdminClient(bootstrap_servers="localhost:9092", client_id='test')

topic_list = []
topic_list.append(NewTopic(name="example_topic", num_partitions=1, replication_factor=1))
admin_client.create_topics(new_topics=topic_list, validate_only=False)

【讨论】：

【解决方案3】：

我发现了这个：

https://github.com/confluentinc/confluent-kafka-python

它对我有用，至少在创建主题时，我指定了一个或多个主题，并在我的 kafka 代理中创建它们。

看我自己的答案：

Could I create a topic in Kafka via API?

我问的和你差不多。

使用示例：

from confluent_kafka.admin import AdminClient, NewTopic

topic = sys.argv[1]

topics = ["newTopicExample","newTopicExample2"]

# Create topic in our Kafka, using kafka-python library.
a = AdminClient({'bootstrap.servers': 'myKafkaBrokerURL'})

new_topics = [NewTopic(topic, num_partitions=3, replication_factor=1) for topic in topics]

# Call create_topics to asynchronously create topics. A dict
# of <topic,future> is returned.
fs = a.create_topics(new_topics)

# Wait for each operation to finish.
for topic, f in fs.items():
    try:
        f.result()  # The result itself is None
        print("Topic {} created".format(topic))
    except Exception as e:
        print("Failed to create topic {}: {}".format(topic, e))

【讨论】：