【问题标题】:Read from specific Kafka topic using Python使用 Python 从特定的 Kafka 主题中读取
【发布时间】:2017-10-22 00:41:35
【问题描述】:

我有 3 个分区的主题,我正在尝试使用以下代码从每个特定分区中读取数据

from kafka import KafkaConsumer, TopicPartition

brokers = 'localhost:9092'
topic = 'b3'

m = KafkaConsumer(topic, bootstrap_servers=['localhost:9092'])
par = TopicPartition(topic=topic, partition=1)
m.assign(par)

但我收到此错误:

    raise IllegalStateError(self._SUBSCRIPTION_EXCEPTION_MESSAGE)
kafka.errors.IllegalStateError: IllegalStateError: You must choose only one way to configure your consumer: (1) subscribe to specific topics by name, (2) subscribe to topics matching a regex pattern, (3) assign itself specific topic-partitions.

有人可以帮我解决这个问题吗?

【问题讨论】:

    标签: python apache-kafka kafka-python


    【解决方案1】:

    你能从 KafkaConsumer() 中删除主题参数并重试吗?

    示例:

    # manually assign the partition list for the consumer
    from kafka import TopicPartition, KafkaConsumer
    consumer = KafkaConsumer(bootstrap_servers='localhost:1234')
    consumer.assign([TopicPartition('foobar', 2)])
    msg = next(consumer)
    

    参考:http://kafka-python.readthedocs.io/en/master/

    【讨论】:

    • 如何在此处插入超时功能?如果没有出现新消息(例如 1s),我想停止流式传输过程。
    • 查看poll 函数,它允许您设置超时。 kafka-python.readthedocs.io/en/master/apidoc/…
    • 顺便说一句:我回答你原来的问题了吗? :-)
    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2016-02-28
    • 2015-08-24
    • 2019-06-24
    • 1970-01-01
    相关资源
    最近更新 更多