kafka-python消费者给出错误

Question

我对kafka和kafka-python相当陌生。安装kafka-python后，我从这里尝试了一个简单的消费者代码实现-http://kafka-python.readthedocs.io/en/master/usage.html

我一直在kafka的bin目录中编写消费者代码，并尝试从那里运行python代码。但是，我遇到以下错误：

回溯（最近一次调用）：文件 “KafkaConsumer.py”，第 4 行，在 for consumer 中的消息中：文件 “/usr/local/lib/python2.7/dist-packages/kafka/vendor/six.py”，第 559 行，在下一个返回类型（self）.next（self） 文件 “/usr/local/lib/python2.7/dist-packages/kafka/consumer/group.py”，第 915 行，在下一个返回 next（self._iterator） 文件 “/usr/local/lib/python2.7/dist-packages/kafka/consumer/group.py”， 第 876 行，_message_generator self._fetcher 中的 msg：文件 “/usr/local/lib/python2.7/dist-packages/kafka/vendor/six.py”，第 559 行，在下一个返回类型（self）.next（self） 文件中 “/usr/local/lib/python2.7/dist-packages/kafka/consumer/fetcher.py”，第 520 行，在下一个返回 next（self._iterator） 文件 “/usr/local/lib/python2.7/dist-packages/kafka/consumer/fetcher.py”，第 477 行， 在 self._unpack_message_set（tp， messages） 中 msg 的_message_generator中：文件 “/usr/local/lib/python2.7/dist-packages/kafka/consumer/fetcher.py”，第 372 行，在 _unpack_message_set inner_mset = msg.decompress（） 文件 “/usr/local/lib/python2.7/dist-packages/kafka/protocol/message.py”，第 121 行，在解压缩断言 has_snappy（），'Snappy 解压缩不受支持' 断言错误：不支持快速解压缩

这是我一直在尝试运行的代码：

from kafka import KafkaConsumer
consumer = KafkaConsumer ('mytopic',bootstrap_servers = ['localhost:9092'], group_id='test-consumer-group')
print "Consuming messages from the given topic"
for message in consumer:
    print("%s:%d%d: key=%s value=%s"  % (message.topic, message.partition, message.offset, message.key, message.value))

因为，我真的是Kafka的新手，我很难理解我做错了什么。

Answer 1

您似乎缺少python-snappy，这是读取以snappy格式压缩的数据所必需的。

您需要< code>snappy和< code>snappy-devel，您可以使用yum、apt-get等安装它们。然后尝试< code > pip install python-snappy