java - 以编程方式合并两个 avro 模式

Question

我有两个相似的模式，其中只有一个嵌套字段发生变化(在 schema1 中称为 onefield，在 schema2 中称为 anotherfield)。

架构 1

{
    "type": "record",
    "name": "event",
    "namespace": "foo",
    "fields": [
        {
            "name": "metadata",
            "type": {
                "type": "record",
                "name": "event",
                "namespace": "foo.metadata",
                "fields": [
                    {
                        "name": "onefield",
                        "type": [
                            "null",
                            "string"
                        ],
                        "default": null
                    }
                ]
            },
            "default": null
        }
    ]
}

架构2

{
    "type": "record",
    "name": "event",
    "namespace": "foo",
    "fields": [
        {
            "name": "metadata",
            "type": {
                "type": "record",
                "name": "event",
                "namespace": "foo.metadata",
                "fields": [
                    {
                        "name": "anotherfield",
                        "type": [
                            "null",
                            "string"
                        ],
                        "default": null
                    }
                ]
            },
            "default": null
        }
    ]
}

我能够使用 avro 1.8.0 以编程方式合并两个模式:

Schema s1 = new Schema.Parser().parse(schema1);
Schema s2 = new Schema.Parser().parse(schema2);
Schema[] schemas = {s1, s2};

Schema mergedSchema = null;
for (Schema schema: schemas) {
    mergedSchema = AvroStorageUtils.mergeSchema(mergedSchema, schema);
}

并使用它将输入的 json 转换为 avro 或 json 表示形式:

JsonAvroConverter converter = new JsonAvroConverter();
try {
    byte[] example = new String("{}").getBytes("UTF-8");
    byte[] avro = converter.convertToAvro(example, mergedSchema);
    byte[] json = converter.convertToJson(avro, mergedSchema);
    System.out.println(new String(json));
} catch (AvroConversionException e) {
    e.printStackTrace();
}

该代码显示了预期的输出:{"metadata":{"onefield":null,"anotherfield":null}}。问题是我看不到合并的架构。如果我执行一个简单的 System.out.println(mergedSchema) 我会得到以下异常:

Exception in thread "main" org.apache.avro.SchemaParseException: Can't redefine: merged schema (generated by AvroStorage).merged
    at org.apache.avro.Schema$Names.put(Schema.java:1127)
    at org.apache.avro.Schema$NamedSchema.writeNameRef(Schema.java:561)
    at org.apache.avro.Schema$RecordSchema.toJson(Schema.java:689)
    at org.apache.avro.Schema$RecordSchema.fieldsToJson(Schema.java:715)
    at org.apache.avro.Schema$RecordSchema.toJson(Schema.java:700)
    at org.apache.avro.Schema.toString(Schema.java:323)
    at org.apache.avro.Schema.toString(Schema.java:313)
    at java.lang.String.valueOf(String.java:2982)
    at java.lang.StringBuilder.append(StringBuilder.java:131)

我称之为 avro 不确定性原理 :)。看起来 avro 能够使用合并的模式，但是当它尝试将模式序列化为 JSON 时失败了。合并适用于更简单的模式，因此对我来说这听起来像是 avro 1.8.0 中的错误。

您知道可能会发生什么或如何解决吗？欢迎任何解决方法(例如:替代 Schema 序列化程序)。

Answer 1

我在 pig util 类中发现了同样的问题......实际上这里有 2 个错误

• AVRO 允许使用无效模式通过 GenericDatumWriter 序列化数据
• piggybank util 类正在生成无效的模式，因为它对所有合并字段使用相同的名称/ namespace (保留原始名称的实例)

这适用于更复杂的场景 https://github.com/kite-sdk/kite/blob/master/kite-data/kite-data-core/src/main/java/org/kitesdk/data/spi/SchemaUtil.java#L511

    Schema mergedSchema = SchemaUtil.merge(s1, s2);

根据您的示例，我得到以下输出

{
  "type": "record",
  "name": "event",
  "namespace": "foo",
  "fields": [
    {
      "name": "metadata",
      "type": {
        "type": "record",
        "name": "event",
        "namespace": "foo.metadata",
        "fields": [
          {
            "name": "onefield",
            "type": [
              "null",
              "string"
            ],
            "default": null
          },
          {
            "name": "anotherfield",
            "type": [
              "null",
              "string"
            ],
            "default": null
          }
        ]
      },
      "default": null
    }
  ]
}

希望这对其他人有帮助。