如何使用 Python decode/deserialize Kafka Avro 字符串

1 回答

• 1

  我认为 Avro 库只是为了阅读 Avro 文件，但它实际上解决了解码 Kafka 消息的问题，如下所示：我首先导入库并将模式文件作为参数，然后创建一个函数将消息解码为字典，我可以在消费者循环中使用。

  import io
  
  from confluent_kafka import Consumer, KafkaError
  from avro.io import DatumReader, BinaryDecoder
  import avro.schema
  
  schema = avro.schema.Parse(open("data_sources/EventRecord.avsc").read())
  reader = DatumReader(schema)
  
  def decode(msg_value):
      message_bytes = io.BytesIO(msg_value)
      decoder = BinaryDecoder(message_bytes)
      event_dict = reader.read(decoder)
      return event_dict
  
  c = Consumer()
  c.subscribe(topic)
  running = True
  while running:
      msg = c.poll()
      if not msg.error():
          msg_value = msg.value()
          event_dict = decode(msg_value)
          print(event_dict)
      elif msg.error().code() != KafkaError._PARTITION_EOF:
          print(msg.error())
          running = False
  

  回复于 2024-04-26T07:00:51+08:00