Python用avro存储库反序列化kafka消息

Question

from kafka import KafkaConsumer
consumer = KafkaConsumer('SOME-TOPIC', 
                        other connection parameters,
                        auto_offset_reset= 'earliest')
                        # value_deserializer=lambda m: json.loads(m.decode('utf-8')))
                        # value_deserializer=lambda m: decode(m))

for msg in consumer:
    print (msg)

null

null

Answer 1

我使用以下库获得了它：

pip install confluent-avro
pip install kafka-python

和代码：

from kafka import KafkaConsumer

from confluent_avro import AvroKeyValueSerde, SchemaRegistry
from confluent_avro.schema_registry import HTTPBasicAuth

KAFKA_TOPIC = "SOME-TOPIC"

registry_client = SchemaRegistry(
    "https://...",
    HTTPBasicAuth("USER", "PASSWORD"),
    headers={"Content-Type": "application/vnd.schemaregistry.v1+json"},
)
avroSerde = AvroKeyValueSerde(registry_client, KAFKA_TOPIC)

consumer = KafkaConsumer(KAFKA_TOPIC, 
                        other connection parameters,
                        auto_offset_reset= 'earliest')

for msg in consumer:
    v = avroSerde.value.deserialize(msg.value)
    k = avroSerde.key.deserialize(msg.key)
    print(msg.offset, msg.partition, k, v)
    break

参考:https://pypi.org/project/confluent_avro/