Avro向后兼容性不像预期的那样工作

Question

我有两个Avro模式V1和V2，在spark中读取如下：

import org.apache.spark.sql.avro.functions._

val jsonFormatSchema = new String(Files.readAllBytes(Paths.get("./examples/src/main/resources/V1.avsc")))

val df = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "topic1")
  .load()

val output = df
  .select(from_avro($"value", jsonFormatSchema) as $"avroFields")

V1有两个字段“一”和“二”

{
  "name": "test",
  "namespace": "foo.bar",
  "type": "record",
  "fields": [
    {
      "name": "one",
      "type": [
        "null",
        "string"
      ],
      "default": null
    },
    {
      "name": "two",
      "type": [
        "null",
        "string"
      ],
      "default": null
    }
  ]
}

V2 与新字段：“三”

{
  "name": "test",
  "namespace": "foo.bar",
  "type": "record",
  "fields": [
    {
      "name": "one",
      "type": [
        "null",
        "string"
      ],
      "default": null
    },
    {
      "name": "two",
      "type": [
        "null",
        "string"
      ],
      "default": null
    },
    {
      "name": "three",
      "type": [
        "null",
        "string"
      ],
      "default": null
    }
  ]
}

场景：编写器使用 V1 进行写入，读取器使用 V2 对 avro 记录进行解码。我的期望是看到字段3填充了默认值，即null。但是我在spark工作中遇到了以下异常。

我是不是错过了什么？我的理解是avro支持向后兼容。

Exception in thread "main" java.io.EOFException
  at org.apache.avro.io.BinaryDecoder.ensureBounds(BinaryDecoder.java:473)
  at org.apache.avro.io.BinaryDecoder.readInt(BinaryDecoder.java:128)
  at org.apache.avro.io.BinaryDecoder.readIndex(BinaryDecoder.java:423)
  at org.apache.avro.io.ResolvingDecoder.doAction(ResolvingDecoder.java:290)
  at org.apache.avro.io.parsing.Parser.advance(Parser.java:88)
  at org.apache.avro.io.ResolvingDecoder.readIndex(ResolvingDecoder.java:267)
  at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:179)
  at org.apache.avro.specific.SpecificDatumReader.readField(SpecificDatumReader.java:116)
  at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:222)
  at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:175)
  at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:153)
  at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:145)

Answer 1

您总是必须使用写入的确切模式来解码Avro。这是因为Avro使用未标记的数据来更紧凑，并且要求编写器模式在解码时出现。

因此，当您使用V2模式阅读时，它会查找字段＜code＞三个＜code＞（或者该字段的空标记），并抛出一个错误。

您可以做的是将解码数据（用writers模式解码）映射到阅读器模式，Java有一个API：＜code＞SpecificDatumReader（模式编写器，模式读取器）＜/code＞。

协议缓冲区或节俭做你想要的，是标记的格式。Avro 希望架构与数据一起传输，例如在 Avro 文件中。