附加分配给特定分区的 KafaConsumer

Question

如何将 python 消费者脚本附加到特定的 kafka 分区。
在运行消费者脚本的两个实例（如下所示）时，每个实例都会随机选择一个分区，然后按预期消费/打印该特定分区的所有消息。

但由于我需要将这些消息输出到磁盘上名为本地文件的分区，因此将脚本的每个实例附加到预先声明的分区 ID 将使事情变得更容易
。：

Date/Hour/PARTITION_ID-0.CSV
Date/Hour/PARTITION_ID-1.CSV

关于如何实现这一目标的任何想法。
请随意提出替代方法。

卡夫卡设置：

Topic:my-topic3 PartitionCount:2    ReplicationFactor:2 Configs:
Topic: my-topic3    Partition: 0    Leader: 2   Replicas: 2,1   Isr: 2,1
Topic: my-topic3    Partition: 1    Leader: 1   Replicas: 1,2   Isr: 1,2

卡夫卡消费者脚本（Python）[带修复]

from kafka import KafkaConsumer
from kafka import TopicPartition

# To consume latest messages and auto-commit offsets
#consumer = KafkaConsumer('my-topic3',
#                         group_id='my-group',
#                         bootstrap_servers=['192.168.150.80:9092'])

# To consume messages from a specific PARTITION  [ FIX ]
consumer = KafkaConsumer(bootstrap_servers='192.168.150.80:9092')
consumer.assign([TopicPartition('my-topic3', 1)])

for message in consumer:
    # message value and key are raw bytes -- decode if necessary!
    # e.g., for unicode: `message.value.decode('utf-8')`
    print ("Topic= %s : Partition= %d : Offset= %d: key= %s value= %s" % (message.topic, message.partition,
                                          message.offset, message.key,
                                          message.value))

更新：如下所示，我使用了分配函数，但不断出现非法状态错误
分配函数

consumer.assign([TopicPartition('my-topic3',1)])

错误

    Traceback (most recent call last):
  File "consumerExample.py", line 13, in <module>
    consumer.assign([TopicPartition('my-topic3',1)])
  File "/usr/lib/python2.7/site-packages/kafka/consumer/group.py", line 278, in assign
    self._subscription.assign_from_user(partitions)
  File "/usr/lib/python2.7/site-packages/kafka/consumer/subscription_state.py", line 189, in assign_from_user
    raise IllegalStateError(self._SUBSCRIPTION_EXCEPTION_MESSAGE)
kafka.errors.IllegalStateError: You must choose only one way to configure

Answer 1

您可以使用该assign()方法手动将一个或多个分区分配给消费者。

这里有一些示例代码：

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