将字节数组转换为Json并给出avro模式作为输入将给出错误 [英] Converting byte array to Json giving avro Schema as input is giving an error

问题描述

我有一个简单的JSON

I have a simple JSON

  String jsonPayload = "{\"empid\": \"6\",\"empname\": \"Saurabh\",\"address\": \"home\"}";
    jsonPayload.getBytes();

我创建了Avro模式

{"namespace": "sample.namespace",
 "type": "record",
 "name": "Employee",
 "fields": [
     {"name": "empid", "type": "string"},
     {"name": "empname",  "type": "string"},
     {"name": "address", "type": "string"}
 ]
}

当我尝试比较它们时，我得到一个错误 例外:

When I try to compare them I get an error Exception :

org.apache.avro.AvroRuntimeException: Malformed data. Length is negative: -62
    at org.apache.avro.io.BinaryDecoder.doReadBytes(BinaryDecoder.java:336)
    at org.apache.avro.io.BinaryDecoder.readString(BinaryDecoder.java:263)
    at org.apache.avro.io.ResolvingDecoder.readString(ResolvingDecoder.java:201)
    at org.apache.avro.generic.GenericDatumReader.readString(GenericDatumReader.java:430)
    at org.apache.avro.generic.GenericDatumReader.readString(GenericDatumReader.java:422)
    at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:180)
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:152)
    at org.apache.avro.generic.GenericDatumReader.readField(GenericDatumReader.java:240)
    at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:230)
    at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:174)
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:152)
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:144)

看起来像String和Charsequence标识有问题.无法确定确切的问题

Looks like there is problem with String and Charsequence identification. Not able to identify exact problem

bytearraytojson转换器方法代码

bytearraytojson converter method code

public String byteArrayToJson(byte[] avro, Schema schema) throws IOException {
        boolean pretty = false;
        GenericDatumReader<GenericRecord> reader = null;
        JsonEncoder encoder = null;
        ByteArrayOutputStream output = null;
        try {

            reader = new GenericDatumReader<GenericRecord>(schema);
            InputStream input = new ByteArrayInputStream(avro);
            output = new ByteArrayOutputStream();
            DatumWriter<GenericRecord> writer = new GenericDatumWriter<GenericRecord>(schema);
            encoder = EncoderFactory.get().jsonEncoder(schema, output, pretty);
            Decoder decoder = DecoderFactory.get().binaryDecoder(input, null);
            GenericRecord datum;
            while (true) {
                try {
                    datum = reader.read(null, decoder);
                } catch (EOFException eofe) {
                    break;
                }
                writer.write(datum, encoder);
            }
            encoder.flush();
            output.flush();
            return new String(output.toByteArray());
        } finally {
            try {
                if (output != null) output.close();
            } catch (Exception e) {
            }
        }
    }

推荐答案

您的问题是avro包含了架构.

Your problem is that the avro has the schema included.

如果要读取Avro，则应使用其他DataReader，DataFileReader

If you want to read the avro you should to use other DataReader, DataFileReader

这里是一个示例，说明如何使用模式读取byte []格式的avro

Here is a example that how read an avro in byte[] format with schema

Scala示例:

def deserializeGenericWithSchema(message: Array[Byte]): Seq[GenericRecord] = {
  val reader: DatumReader[GenericRecord] = new SpecificDatumReader[GenericRecord]()
  val fileReader = new DataFileReader(new SeekableByteArrayInput(message),reader)
  extractRec(fileReader,Seq.empty[GenericRecord])
}

@tailrec
def extractRec(fileReader: DataFileReader[GenericRecord], acc: Seq[GenericRecord]):Seq[GenericRecord] = {
  if (fileReader.hasNext) {
    val newElement = fileReader.next
    extractRec(fileReader,acc :+ newElement)
  } else {
    acc
  }
}

Java示例:

public List<GenericRecord> deserializeGenericWithSchema(byte[] message) throws IOException {
    List<GenericRecord>listOfRecords = new ArrayList<>();
    DatumReader<GenericRecord> reader = new SpecificDatumReader<>();
    DataFileReader<GenericRecord> fileReader =
            new DataFileReader<>(new SeekableByteArrayInput(message),reader);
    while (fileReader.hasNext()) {
        listOfRecords.add(fileReader.next());
    }
    return listOfRecords;
}

PD:我已经在Scala中编写了解决方案，然后未经测试就接触了Java.也许Java解决方案并不完全完美

PD: I have written the solution in scala and then I have traduced to Java, without testing. Maybe the Java solution is not completely perfect

这篇关于将字节数组转换为Json并给出avro模式作为输入将给出错误的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！