GenericRecord的Apache波束编码器

Question

我正在构建一个读取Avro通用记录的管道。为了在各个阶段之间传递GenericRecord，我需要注册avrocoder。文档表明，如果我使用通用记录，模式参数可以是任意的:https://beam.apache.org/releases/javadoc/2.2.0/org/apache/beam/sdk/coders/avrocoder.html#of-java.lang.class-org.apache.avro.schema-

但是，当我将空模式传递给方法avrocoder.of(Class,schema)时，它会在运行时引发异常。有没有一种方法可以为GenericRecord创建一个不需要模式的AvroCoder？在我的例子中，每个GenericRecord都有一个嵌入的模式。

异常和StackTrace：

Exception in thread "main" java.lang.NullPointerException
at org.apache.beam.sdk.coders.AvroCoder$AvroDeterminismChecker.checkIndexedRecord(AvroCoder.java:562)
at org.apache.beam.sdk.coders.AvroCoder$AvroDeterminismChecker.recurse(AvroCoder.java:430)
at org.apache.beam.sdk.coders.AvroCoder$AvroDeterminismChecker.check(AvroCoder.java:409)
at org.apache.beam.sdk.coders.AvroCoder.<init>(AvroCoder.java:260)
at org.apache.beam.sdk.coders.AvroCoder.of(AvroCoder.java:141)

Answer 1

在查看了avrocoder的代码之后，我认为这里的文档并不正确。您的avrocoder实例将需要一种方法来确定Avro记录的模式-可能唯一的方法是提供一种方法。

因此，我建议调用avrocoder.of(GenericRecord.class,schema)，其中schema是pCollection中GenericRecord对象的正确模式。