How to programmatically create a topic in Apache Kafka using Python

Question

So far I haven't seen a python client that implements the creation of a topic explicitly without using the configuration option to create automatically the topics.

Answer 1

You can programmatically create topics using either kafka-python or confluent_kafka client which is a lightweight wrapper around librdkafka.

---

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

---

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

Answer 2

If you can run confluent_kafka (Python) v0.11.6 or above, then the following is how to create kafka topics, list kafka topics and 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)}

And to delete kafka topics using that same kafka_admin object, this:

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

I hope this helps. \(◠﹏◠)/

Answer 3

It looks like you can can use the following to ensure that your topic already exist (I assume you are using the following kafka python implementation):

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

Answer 4

It seems that there is no kafka server api to create a topic so you have to use topic automatic creation of the or the command line tool:

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

Answer 5

It's already too late. I don't know about a command for explicit creation of the topics but the following creates and adds the messages.

I created a python kafka producer:

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

In the list of Kafka topics xyz was not there previously. when I did the above method, the Python-kafka client created it and added the messages to it.

Answer 6

from kafka import KafkaProducer

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

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

Answer 7

The AdminClient API needed to do programmatic topic creation and configuration was just added in Kafka 0.11 (initially for Java)

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

It is expected that non-Java client libraries will add this functionality as well over time. Check with the author of the Kafka Python client you are using (there are several) to see if and when KIP-4 admin protocol support will be in the API

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