42

到目前为止，我还没有看到一个 python 客户端在不使用配置选项自动创建主题的情况下显式实现主题的创建。

4

7 回答 7

61

您可以使用kafka-python或confluent_kafkaclient 以编程方式创建主题，它是librdkafka的轻量级包装器。

使用kafka-python

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)

使用confluent_kafka

from confluent_kafka.admin import AdminClient, NewTopic


admin_client = AdminClient({
    "bootstrap.servers": "localhost:9092"
})

topic_list = []
topic_list.append(NewTopic("example_topic", 1, 1))
admin_client.create_topics(topic_list)
于 2019-04-04T20:39:47.417 回答
7

如果您可以运行confluent_kafka（Python）v0.11.6或更高版本，那么以下是如何运行create kafka topics，list kafka topics和delete kafka topics：

>>> import confluent_kafka.admin, pprint

>>> conf        = {'bootstrap.servers': 'broker01:9092'}
>>> kafka_admin = confluent_kafka.admin.AdminClient(conf)

>>> new_topic   = confluent_kafka.admin.NewTopic('topic100', 1, 1)
                  # Number-of-partitions  = 1
                  # Number-of-replicas    = 1

>>> kafka_admin.create_topics([new_topic,]) # CREATE (a list(), so you can create multiple).
    {'topic100': <Future at 0x7f524b0f1240 state=running>} # Stdout from above command.

>>> pprint.pprint(kafka_admin.list_topics().topics) # LIST
    {'topic100' : TopicMetadata(topic100, 1 partitions),
     'topic99'  : TopicMetadata(topic99,  1 partitions),
     'topic98'  : TopicMetadata(topic98,  1 partitions)}

并delete kafka topics使用相同的kafka_admin对象，这：

kafka_admin.delete_topics(['topic99', 'topic100',]) # DELETE

我希望这有帮助。\(◠﹏◠)/

于 2019-01-24T17:47:49.473 回答
2

看起来您可以使用以下内容来确保您的主题已经存在（我假设您正在使用以下kafka python实现）：

client = KafkaClient(...)
producer = KafkaProducer(...)
client.ensure_topic_exists('my_new_topic')
producer.send_messages('my_new_topic', ...)
于 2015-01-26T05:09:17.383 回答
1

似乎没有kafka server api来创建主题，所以您必须使用主题自动创建或命令行工具：

bin/kafka-create-topic.sh --zookeeper localhost:2181 --replica 1 --partition 1 --topic test
于 2014-09-25T11:50:20.470 回答
0 
from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers=['localhost:9092'])
topic = 'topic-name'

producer.send(topic, final_list[0]).get(timeout=10)
于 2018-03-10T17:42:10.150 回答
0

进行编程主题创建和配置所需的 AdminClient API 刚刚添加到 Kafka 0.11（最初用于 Java）

见https://cwiki.apache.org/confluence/display/KAFKA/KIP-117%3A+Add+a+public+AdminClient+API+for+Kafka+admin+operations

预计非 Java 客户端库也会随着时间的推移添加此功能。请咨询您正在使用的 Kafka Python 客户端的作者（有几个），了解 API 中是否以及何时支持 KIP-4 管理协议

见https://cwiki.apache.org/confluence/display/KAFKA/KIP-4+-+Command+line+and+centralized+administrative+operations

于 2017-06-21T06:20:56.683 回答
0

已经太晚了。我不知道用于显式创建主题的命令，但以下创建并添加消息。

我创建了一个 python kafka 生产者：

prod = KafkaProducer(bootstrap_servers='localhost:9092')
for i in xrange(1000):
    prod.send('xyz', str(i))

在 Kafka 主题列表中xyz以前不存在。当我执行上述方法时，Python-kafka 客户端创建了它并将消息添加到它。

于 2017-06-20T08:28:43.893 回答