Я начинаю работать с Kafka и довольно новичок в Python. Я использую эту библиотеку с именем kafka-python для связи с моим Kafka. маклер. Теперь мне нужно динамически создать тему из моего кода, из документов я вижу, что я могу вызвать метод create_topics() для этого, однако я не уверен, как я получу экземпляр этого класса. Я не могу понять это из документов.
Может кто-то помочь мне с этим?
Сначала вам нужно создать экземпляр KafkaAdminClient< /а>. Следующее должно помочь вам:
from kafka.admin import KafkaAdminClient, NewTopic
admin_client = KafkaAdminClient(
bootstrap_servers="localhost:9092",
client_id='test'
)
topic_list = [NewTopic(name="example_topic", num_partitions=1, replication_factor=1)]
admin_client.create_topics(new_topics=topic_list, validate_only=False)
Кроме того, вы можете использовать клиент confluent_kafka, который представляет собой облегченную оболочку для librdkafka:
from confluent_kafka.admin import AdminClient, NewTopic
admin_client = AdminClient({"bootstrap_servers": "localhost:9092"})
topic_list = [NewTopic("example_topic", 1, 1)]
admin_client.create_topics(topic_list)
