【问题标题】:How to check number of partitions in Kafka(confluent_kafka)如何检查 Kafka 中的分区数(confluent_kafka)
【发布时间】:2021-08-09 08:41:17
【问题描述】:

我正在尝试通过在 0 0 * * * 每天使用 confluent_kafka Python 包来进行批处理 etl。我知道我的流中有 4 个分区,但它可以更改,所以有没有办法检查特定主题中的分区总数? 我的消费者就是这样;

from confluent_kafka import Consumer, KafkaError

    messages = list()
    partition_counter = 0
    tnof_partition = 4

    while True:
        msg = self.consumer.poll(0.1)
        if msg is None:
            continue
        elif not msg.error():
            event = json.loads(msg.value().decode('utf-8'))

        elif msg.error().code() == KafkaError._PARTITION_EOF:
            print("End of partition reached {0}/{1}"
                .format(msg.topic(), msg.partition()))
            
            partition_counter += 1
            if(partition_counter == tnof_partition):
                self.consumer.commit()
                self.consumer.close()
                break

如果您能展示实现批量消费者的替代方法,我将不胜感激。谢谢

【问题讨论】:

  • “但可以更改” - 然后启用身份验证并禁止管理员客户端修改,那么您就不需要此断言

标签: python apache-kafka consumer confluent-kafka-python


【解决方案1】:

消费者的list_topics()方法可以提供由TopicMetadata组成的Topics的映射,其中最终包含partitions

参考:https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#confluent_kafka.Consumer.list_topics

【讨论】:

    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 2018-10-14
    • 1970-01-01
    • 1970-01-01
    • 2019-01-24
    • 2018-06-26
    • 2018-10-08
    • 2018-01-11
    相关资源
    最近更新 更多