如何为Apache Beam中的Avro SpecificRecordBase的所有子类选择单个编码器？

问题描述 投票：0回答：1

背景

我的光束管道被设计为处理Avro SpecificRecordBase类型的元素。

为了简化我的问题，假设我有两种以Avro格式生成的元素，它们都有各自的字段：

class Dog extends SpecificRecordBase {
    ....
}

class Cat extends SpecificRecordBase {
   ...
}

管道将从输入的Kafka中读取元素，处理元素并将已处理的元素放入输出Kafka中，如下所示：

Pipeline pipeline = Pipeline.create(getOptions());
pipeline.getCoderRegistry().registerCoderForClass(SpecificRecordBase.class, <what shall I put here?>);

pipeline.apply(kafkaReaderTransformer)
              .apply(Window.into(FixedWindows.of(Duration.standardSeconds(getWindowSize()))))
                .apply(GroupByKey.create())
                .apply(ParDo.of(GiveShowerToPetDoFn))
                .apply(Flatten.iterables())
                .apply(kafkaWriterTransformer);

问题

我的问题是如何在管道中注册编码器？由于将来可以从Cat Kafka或Dog Kafka甚至Toad Kafka中读取pipleline，因此我需要一种通用的方式来注册编码器，该编码器可以序列化在运行时确定的SpecificRecordBase的所有子类。

我的失败解决方案

我尝试了以下代码来填充代码中的空白：

  1. AvroCoder.of（SpecificRecordBase.class）：不起作用

    我在运行管道时出现以下错误：

Caused by: avro.shaded.com.google.common.util.concurrent.UncheckedExecutionException: org.apache.avro.AvroRuntimeException: Not a Specific class: class org.apache.avro.specific.SpecificRecordBase
 at avro.shaded.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2234)
 at avro.shaded.com.google.common.cache.LocalCache.get(LocalCache.java:3965)
 at avro.shaded.com.google.common.cache.LocalCache.getOrLoad(LocalCache.java:3969)
 at avro.shaded.com.google.common.cache.LocalCache$LocalManualCache.get(LocalCache.java:4829)
 at org.apache.avro.specific.SpecificData.getSchema(SpecificData.java:225)
 ... 23 more
Caused by: org.apache.avro.AvroRuntimeException: Not a Specific class: class org.apache.avro.specific.SpecificRecordBase
 at org.apache.avro.specific.SpecificData.createSchema(SpecificData.java:285)
 at org.apache.avro.reflect.ReflectData.createSchema(ReflectData.java:594)
 at org.apache.avro.specific.SpecificData$2.load(SpecificData.java:218)
 at org.apache.avro.specific.SpecificData$2.load(SpecificData.java:215)
 at avro.shaded.com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3568)
 at avro.shaded.com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2350)
 at avro.shaded.com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(LocalCache.java:2313)
 at avro.shaded.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2228)
 ... 27 more

  1. SerializableCoder.of（SpecificRecordBase.class）：引发令人困惑的异常

    这应该是一个有前途的选择，但是当我运行Pipeline时，我在下面遇到了一个非常令人困惑的错误，以下内容令人困惑，因为Cat实际上是通过继承自SpecificRecordBase实现可序列化的：

   Caused by: java.lang.ClassCastException: Cat cannot be cast to java.io.Serializable
    at org.apache.beam.sdk.coders.SerializableCoder.encode(SerializableCoder.java:53)
    at org.apache.beam.sdk.coders.Coder.encode(Coder.java:136)
    at org.apache.beam.sdk.util.WindowedValue$FullWindowedValueCoder.encode(WindowedValue.java:578)
    at org.apache.beam.sdk.util.WindowedValue$FullWindowedValueCoder.encode(WindowedValue.java:569)
    at org.apache.beam.sdk.util.WindowedValue$FullWindowedValueCoder.encode(WindowedValue.java:529)
    at org.apache.beam.runners.spark.coders.CoderHelpers.toByteArray(CoderHelpers.java:53)
    at org.apache.beam.runners.spark.coders.CoderHelpers.lambda$toByteFunction$28e77fe8$1(CoderHelpers.java:143)
    at org.apache.spark.api.java.JavaPairRDD$$anonfun$pairFunToScalaFun$1.apply(JavaPairRDD.scala:1043)
    at org.apache.spark.api.java.JavaPairRDD$$anonfun$pairFunToScalaFun$1.apply(JavaPairRDD.scala:1043)
    at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
    at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:149)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:96)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53)
    at org.apache.spark.scheduler.Task.run(Task.scala:109)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)
  1. 不要自己严格编码器，让Beam推断。系统将为我推断合适的编码器。这种无能为力的解决方案在我的Loal独立机器上无法工作，但是当我将它们放入真正的多服务器环境中时，它们只是抛出异常，表明它们无法推断编码器。 
Caused by: java.lang.IllegalStateException: Unable to return a default Coder for ParDo(Deserialize)/ParMultiDo(Deserialize).output [PCollection]. Correct one of the following root causes:
  No Coder has been manually specified;  you may do so using .setCoder().
  Inferring a Coder from the CoderRegistry failed: Cannot provide coder for parameterized type org.apache.beam.sdk.values.KV<java.lang.String, org.apache.avro.specific.SpecificRecordBase>: Unable to provide a Coder for org.apache.avro.specific.SpecificRecordBase.
  Building a Coder using a registered CoderProvider failed.
  See suppressed exceptions for detailed failures.
  Using the default output Coder from the producing PTransform failed: PTransform.getOutputCoder called.
    at org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkState(Preconditions.java:507)
    at org.apache.beam.sdk.values.PCollection.getCoder(PCollection.java:278)
    at org.apache.beam.sdk.values.PCollection.finishSpecifying(PCollection.java:115)
    at org.apache.beam.sdk.runners.TransformHierarchy.finishSpecifyingInput(TransformHierarchy.java:191)
    at org.apache.beam.sdk.Pipeline.applyInternal(Pipeline.java:538)
    at org.apache.beam.sdk.Pipeline.applyTransform(Pipeline.java:473)
    at org.apache.beam.sdk.values.PCollection.apply(PCollection.java:357)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.springframework.beans.factory.support.SimpleInstantiationStrategy.instantiate(SimpleInstantiationStrategy.java:154)
apache-spark avro apache-beam serializable encoder
1个回答
0
投票

我终于使用解决方法解决了此处发布的问题。

根本原因

原来是由不同环境中的编码器不兼容引起的。尽管编码器在我的本地环境中工作，但是prod依赖版本有所不同，导致Beam库无法对从SpecificRecordBase派生的类进行编码和解码。

两种解决方案

1）使用字节作为输入并使用字节作为输出来更改管道中的每个doFun：

public class GiveShowerToPetDoFn extends DoFn<KV<String, byte[]>, KV<String, byte[]>> {
   ...
}

这意味着您将在执行实际的业务逻辑之前手动从字节反序列化对象，并在最后一步将结果序列化回字节。这使Beam使用adoli的doFun之间的默认字节编码器/解码器，并且该编码器/解码器将始终有效，因为它正在处理基本类型而不是自定义类型。

2）为您的自定义类型编写自己的编码器/解码器。

解决方案1和2本质上是相同的。就我而言，我使用第一个解决方案来解决我的问题。

最新问题
© www.soinside.com 2019 - 2024. All rights reserved.