Проблема:
Я столкнулся с ошибкой при попытке использовать оконную обработку с GenericRecord Avro в Apache Beam. Я получаю следующую ошибку:
Фрагмент кода:
Вот код, который вызывает ошибку:
PCollection windowedOutput = inputRecords
.apply(Window.into(FixedWindows.of(Duration.standardMinutes(1))))
.setCoder(KvCoder.of(
KvCoder.of(
AvroCoder.of(GenericRecord.class, keySchema),
AvroCoder.of(GenericRecord.class, valueSchema)
),
BooleanCoder.of()
));
При применении возникает исключение AvroRuntimeException, упомянутое выше.
Рабочий пример:
следующий аналогичный код (без окон) работает нормально:
PCollection outputRecords = input
.apply(ParDo.of(new ConvertToAvroFn(keySchema, valueSchema)))
.setCoder(KvCoder.of(
AvroCoder.of(GenericRecord.class, keySchema),
AvroCoder.of(GenericRecord.class, valueSchema)
));
[ERROR] Encountered an exception, shutting down the process {}
org.apache.avro.AvroRuntimeException: Not a Specific class: interface org.apache.avro.generic.GenericRecord
at org.apache.avro.specific.SpecificData.createSchema(SpecificData.java:396) ~[avro-1.10.2.jar:1.10.2]
at org.apache.avro.reflect.ReflectData.createSchema(ReflectData.java:726) ~[avro-1.10.2.jar:1.10.2]
at org.apache.avro.specific.SpecificData$3.computeValue(SpecificData.java:328) ~[avro-1.10.2.jar:1.10.2]
at java.lang.ClassValue.getFromHashMap(ClassValue.java:227) ~[?:1.8.0_282]
at java.lang.ClassValue.getFromBackup(ClassValue.java:209) ~[?:1.8.0_282]
at java.lang.ClassValue.get(ClassValue.java:115) ~[?:1.8.0_282]
at org.apache.avro.specific.SpecificData.getSchema(SpecificData.java:339) ~[avro-1.10.2.jar:1.10.2]
at org.apache.beam.sdk.coders.AvroCoder.of(AvroCoder.java:163) ~[beam-sdks-java-core-2.45.22.jar:?]
at com.example.pipeline.CustomSink.applyCustomCheck(CustomSink.java:247)
at com.example.pipeline.CustomSink.expand(CustomSink.java:219)
at org.apache.beam.sdk.Pipeline.applyInternal(Pipeline.java:551)
Подробнее здесь: https://stackoverflow.com/questions/789 ... ow-can-i-r
