如何在 Spark 代码中设置 Kryo 的不可修改的集合序列化器

Question

我在 Java 的 Spark (v1.6.1) 中使用 Kryo 序列化，在序列化一个在其字段中有集合的 class 时，它抛出以下错误 -

Caused by: java.lang.UnsupportedOperationException
         at java.util.Collections$UnmodifiableCollection.add(Collections.java:1055)
         at com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:102)
         at com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:18)
         at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:648)
         at com.esotericsoftware.kryo.serializers.FieldSerializer$ObjectField.read(FieldSerializer.java:605)
         ... 27 more

我发现这是因为 Kryo 的默认 CollectionSerializer 不能反序列化集合，因为它是不可修改的，我们应该改用 UnmodifiableCollectionsSerializer。

如何在 spark 代码中特别提及使用 UnmodifiableCollectionsSerializer for Kryo？

我目前的配置是-

SparkConf conf = new SparkConf().setAppName("ABC");
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
conf.registerKryoClasses(new Class<?>[] {*list of classes I want to register*});

Answer 1

万一其他人遇到这个问题，这里是解决方案 - 我通过使用 javakaffee kryo 序列化程序让它工作。

添加如下maven依赖：

<dependency>
        <groupId>de.javakaffee</groupId>
        <artifactId>kryo-serializers</artifactId>
        <version>0.42</version>
</dependency>

编写一个自定义的 kryo 注册器来注册 UnmodifiableCollectionsSerializer

    public class CustomKryoRegistrator implements KryoRegistrator {
        @Override
        public void registerClasses(Kryo kryo) {        
             UnmodifiableCollectionsSerializer.registerSerializers(kryo);
        }
   }

将 spark.kryo.registrator 设置为自定义注册器的完全限定名称

conf.set("spark.kryo.registrator", "com.abc.CustomKryoRegistrator");

参考文献 -

https://github.com/magro/kryo-serializers

如何在 Spark 代码中设置 Kryo 的不可修改的集合序列化器

How to set Unmodifiable collection serializer of Kryo in Spark code

kryo

apache-spark