如何使用Apache Kafka修复Python2.7中的"AssertionError:Value必须为字节"错误

Question

我是Apache Kafka技术的新手.我试图通过使用python 2.7将消息作为JSON对象发送到kafka主题但我得到"AssertionError:Value必须是字节"错误.我可以成功地将消息作为字符串发送,我可以使用kafka-console-consumer.sh查看我的消息.我正在使用apache kafka 2.10-0.8.2.1版本.我在下面给出了我的代码.

from kafka import KafkaProducer
import yaml

producer = KafkaProducer(bootstap_servers="localhost:9092")
msg = yaml.safe_load('{"id":1, "name":"oguz"}')

producer.send("my-topic", msg)

谢谢你的帮助.

Answer 1

yaml.safe_load() 返回一个dict,因此将它转换为字节需要两件事 - 通过JSON将其序列化为字符串,然后将其编码为UTF-8字节.

从kafka-python文档中的示例中,您可以value_serializer在实例化时使用关键字参数KafkaProducer:

>>> import json
>>> producer = KafkaProducer(value_serializer=lambda v: json.dumps(v).encode('utf-8'))
>>> producer.send('my-topic', msg)

或者,您可以在调用时手动将其序列化send():

>>> producer.send('my-topic', json.dumps(msg).encode('utf-8'))