google-cloud-platform - 将 Avrocoder 用于具有泛型的自定义类型

Question

我正在尝试使用 AvroCoder 序列化在我的管道中的 PCollections 中传递的自定义类型。自定义类型有一个通用字段(目前是一个字符串)当我运行管道时，我得到如下所示的 AvroTypeException 可能是由于通用字段。为对象构建和传递 AvroSchema 是解决此问题的唯一方法吗？

Exception in thread "main" org.apache.avro.AvroTypeException: Unknown type: T
 at org.apache.avro.specific.SpecificData.createSchema(SpecificData.java:255)
 at org.apache.avro.reflect.ReflectData.createSchema(ReflectData.java:514)
 at org.apache.avro.reflect.ReflectData.createFieldSchema(ReflectData.java:593)
 at org.apache.avro.reflect.ReflectData.createSchema(ReflectData.java:472)
 at org.apache.avro.specific.SpecificData.getSchema(SpecificData.java:189)
 at com.google.cloud.dataflow.sdk.coders.AvroCoder.of(AvroCoder.java:116)

我还附上了我的注册表代码以供引用。

pipelineCoderRegistry.registerCoder(GenericTypeClass.class, new CoderFactory() {
    @Override
    public Coder<?> create(List<? extends Coder<?>> componentCoders) {
        return AvroCoder.of(GenericTypeClass.class);
    }

    @Override
    public List<Object> getInstanceComponents(Object value) {
        return Collections.singletonList(((GenericTypeClass<Object>) value).key);
    }
});

Answer 1

就设置 CoderFactory 而言，您做的一切都是正确的。 ，但 Avro 的 ReflectData 机制 AvroCoder 在撰写本文时，用于自动生成模式的方法不适用于泛型类型。这被跟踪为问题 AVRO-1571 .另见 this StackOverflow question .

为了允许对 GenericTypeClass<T> 进行编码对于 T 的某些特定值，你是正确的，你必须提供一些明确的模式信息。有两种方法可以继续:

第一种方法是为 T 类型的字段提供显式模式。在您的GenericTypeClass<T> ，像这样:

class GenericTypeClass<T> {
  // Avro requires a no-args constructor
  public GenericTypeClass() {}

  @AvroSchema("[\"string\", \"int\", ...]")
  private T genericField;
}

缺点是它仅限于有限的静态联合模式，并且需要手动内联 JSON 模式以获得更复杂的 T 值。 .

第二种方法是在构建 AvroCoder 时提供显式模式。在您的 CoderFactory ，并将此架构提供给 AvroCoder.of(Class, Schema) .

pipelineCoderRegistry.registerCoder(GenericTypeClass.class, new CoderFactory() {
  @Override
  public Coder<?> create(List<? extends Coder<?>> componentCoders) {
      return AvroCoder.of(
          GenericTypeClass.class
          schemaFromCoder(componentCoders.get(0)));
  }

  ...
});

这将主要围绕转换 Coder<T>进入 T 的模式.这对于基本类型来说应该很容易，对于 ReflectData 的 POJO 来说应该是易于管理的。支持。它还提供了一个钩子(Hook)，用于对更困难的案例进行临时支持。