将字节数组转换为Json，将avro Schema作为输入，会产生错误

Question

我有一个简单的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"}
 ]
}

当我尝试比较它们时，我得到一个错误异常：

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标识有问题。无法确定确切的问题

bytearraytojson转换器方法代码

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

Answer 1

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

如果你想阅读avro，你应该使用其他DataReader，DataFileReader

下面是一个如何使用schema读取byte []格式的avro的示例

示例比例：

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解决方案并不完美

Answer 2

你必须使用morphline将json转换为avro。

这是链接。 http://cloudera.github.io/cdk/docs/current/cdk-morphlines/morphlinesReferenceGuide.html#/cdk-morphlines-avro