时间戳

首页 > 解决方案 > 在一个文件中反序列化一组 Confluent 编码的 Avro

scala - 在一个文件中反序列化一组 Confluent 编码的 Avro

问题描述

我有一个文件，它的二进制 avro 彼此相邻。我想一一阅读每条记录。同时，我想从每个包含模式 id 的记录中读取前几个字节，然后对其进行反序列化。我可以使用下面的代码跳过这些字节，并使用固定模式。这个对我有用。但我想逐一阅读。这可能吗？

  val client = new SchemaRegistryClient("SCHEMA_REGISTRY_URL")
  val schema = new Schema.Parser().parse(client.getSchema("TOPIC_NAME").get.toString)
  val reader = new GenericDatumReader[GenericRecord](schema)
  val filename = "MY_BINARY_AVRO.avro"
  var fileContInBytes = Files.readAllBytes(Paths.get(filename))
  val decoder = DecoderFactory.get.binaryDecoder(fileContInBytes, null)
  while (!decoder.isEnd) {
    decoder.skipFixed(5)
    val rec = reader.read(null, decoder)
  }

能够反序列化二进制 avro、彼此相邻并无缝移动字节位置的 Python 代码

from avro import schema, datafile, io
import io
import avro
import requests
import os

topic=r'TOPIC_NAME'
schemaurl=r'SCHEMA_REGISTRY_URL'
OUTFILE_NAME = r'INPUT_BINARY_AVRO_FILE_LOCATION'
f=open(OUTFILE_NAME,'rb')
buf = io.BytesIO(f.read())
decoder = avro.io.BinaryDecoder(buf)

while buf.tell()<os.path.getsize(OUTFILE_NAME):
  id=int.from_bytes((buf.read(4)), byteorder='big')
  SCHEMA = avro.schema.Parse(getSchema(schemaurl,id))
  rec_reader = avro.io.DatumReader(SCHEMA)
  out=rec_reader.read(decoder)
  print(out)

标签： scalaapache-kafkaavroconfluent-platformconfluent-schema-registry

解决方案

推荐阅读