Как программно создать тему в Apache Kafka с использованием Python

Ответ 1

Вы можете программно создавать темы, используя клиент kafka kafka-python или confluent_kafka который представляет собой облегченную оболочку вокруг 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)

Ответ 2

Если вы можете запустить confluent_kafka (Python) v0.11.6 или выше, то ниже описано, как создавать темы kafka, перечислять темы kafka и удалять темы kafka:

>>> 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)}

И чтобы удалить темы kafka_admin, используя тот же объект kafka_admin, это:

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

Я надеюсь, что эти операции помогут.

Ответ 3

Кажется, что нет kafka server api для создания темы, поэтому вам нужно использовать автоматическое создание темы или инструмента командной строки:

bin/kafka-create-topic.sh --zookeeper localhost:2181 --replica 1 --partition 1 --topic test

Ответ 4

Похоже, вы можете использовать следующее, чтобы убедиться, что ваша тема уже существует (я предполагаю, что вы используете следующую kafka python)

client = KafkaClient(...)
producer = KafkaProducer(...)
client.ensure_topic_exists('my_new_topic')
producer.send_messages('my_new_topic', ...)

Ответ 5

Это уже слишком поздно. Я не знаю о команде для явного создания тем, но следующее создает и добавляет сообщения.

Я создал python kafka продюсер:

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

В списке тем Kafka xyz ранее не было. когда я сделал вышеуказанный метод, клиент Python-kafka создал его и добавил к нему сообщения.

Ответ 6

API-интерфейс AdminClient, необходимый для создания и настройки программных тем, был добавлен в 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 в API

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

Ответ 7

from kafka import KafkaProducer

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

producer.send(topic, final_list[0]).get(timeout=10)