使用AVRO的Python Kafka消费者消息反序列化，没有模式注册-问题

Question

我有一个Kafka消息反序列化的问题。我使用confluent kafka。

没有模式注册表-模式是硬编码的。

我可以将消费者连接到任何主题并接收消息，但我不能反序列化这些消息。

反序列化后的输出如下所示:

打印(读者):

<avro.io.DatumReader object at 0x000002354235DBB0>

我想，我反序列化的代码是错误的，但是如何解决这个问题?

最后我想提取反序列化的键和值

from confluent_kafka import Consumer, KafkaException, KafkaError
import sys
import time
import avro.schema
from avro.io import DatumReader, DatumWriter
def kafka_conf():
conf = {''' MY CONFIGURATION'''
}
return conf

if __name__ == '__main__':
conf = kafka_conf()
topic = """MY TOPIC"""
c = Consumer(conf)
c.subscribe([topic])
try:
while True:
msg = c.poll(timeout=200.0)
if msg is None:
continue
if msg.error():
# Error or event
if msg.error().code() == KafkaError._PARTITION_EOF:
# End of partition event
sys.stderr.write('%% %s [%d] reached end at offset %dn' %
(msg.topic(), msg.partition(), msg.offset()))
else:
# Error
raise KafkaException(msg.error())
else:
print("key: ", msg.key())
print("value: ", msg.value())
print("offset: ", msg.offset())
print("topic: ", msg.topic())
print("timestamp: ", msg.timestamp())
print("headers: ", msg.headers())
print("partition: ", msg.partition())
print("latency: ", msg.latency())
schema = avro.schema.parse(open("MY_AVRO_SCHEMA.avsc", "rb").read())
print(schema)
reader = DatumReader(msg.value, reader_schema=schema)
print(reader)
time.sleep(5)  # only on test
except KeyboardInterrupt:
print('nAborted by usern')
finally:
c.close()

Answer 1

您正在打印一个阅读器对象，而不是反序列化数据，这是您使用reader.read()

你还需要一个BinaryDecoder。

Confluent库源代码中的DeserializingConsumer在从注册表而不是本地文件系统获取模式之后做了完全相同的事情，所以我建议您按照它们的做法去做。