在Beam Spark runner中注册Kryo序列化的自定义类

Question

我已经看到Beam Spark runner使用BeamSparkRunnerRegistrator进行kryo注册。有没有办法注册自定义用户类？

Answer 1

如果您不想注册您的类，Kryo序列化仍然可以工作，但它必须与每个对象一起存储完整的类名，这是浪费。

Answer 2

使用此自定义序列化程序创建您自己的Kryoynstrator

package Mypackage
class MyRegistrator extends KryoRegistrator {
  override def registerClasses(kryo: Kryo) {
    kryo.register(classOf[A], new CustomASerializer())
}}

然后，使用注册者的完全限定名称添加关于它的配置条目，例如Mypackage。我的注册人：

val conf = new SparkConf()
conf.set("spark.kryo.registrator", "Mypackage.KryoRegistrator")

请参阅留档：数据序列化火花

Answer 3

有一个方法可以做到这一点，但是首先，我可以问一下你为什么要这么做吗？

一般来说，Beam的Spark运行器使用Beam编码器来序列化用户数据。

我们目前有一个错误，其中缓存的DStream正在使用Kryo进行序列化，如果用户类不可Kryo序列化，则失败。BEAM-2669。我们目前正在尝试解决此问题。

如果这是您面临的问题，您目前可以使用Kryo的注册器来解决此问题。这是您面临的问题吗？或者您有其他原因这样做，请告诉我。

在任何情况下，以下是您如何使用SparkContextOptions将自己的自定义JavaSparkContext实例提供给Beam的Spark运行程序

SparkConf conf = new SparkConf();
conf.set("spark.serializer", KryoSerializer.class.getName());
conf.set("spark.kryo.registrator", "my.custom.KryoRegistrator");

JavaSparkContext jsc = new JavaSparkContext(..., conf);

SparkContextOptions options = PipelineOptionsFactory.as(SparkContextOptions.class);
options.setRunner(SparkRunner.class);
options.setUsesProvidedSparkContext(true);
options.setProvidedSparkContext(jsc);

Pipeline p = Pipeline.create(options);

有关更多信息，请参阅：

Beam Spark runner文档

示例：ProvidedSparkContextTest。Java语言