Десериализация набора Confluent-закодированного Avro в одном файле - PullRequest
Новая
       64

Десериализация набора Confluent-закодированного Avro в одном файле

0 голосов
/ 13 октября 2018

У меня есть файл, в котором двоичный файл avro добавлен рядом друг с другом.Я хотел бы прочитать каждую запись одну за другой.В то же время я хотел бы прочитать первые несколько байтов из каждой записи, содержащей идентификатор схемы, а затем десериализовать ее.Я могу пропустить эти байты, используя приведенный ниже код, и использовать фиксированную схему.Меня устраивает.Но я хотел бы прочитать каждый в отдельности.Это возможно? 

  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)
  }

Python-код, который способен десериализовать двоичный код avro, присутствовать рядом друг с другом и плавно перемещать позиции байтов 

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)
...