哈希图的火花序列化问题答案

【问题标题】：spark serialization issue with hashmap哈希图的火花序列化问题
【发布时间】：2015-10-15 18:40:55
【问题描述】：

我正在尝试序列化一个 Java 类，其对象将在 Java Spark RDD 中收集。 POJO 包含一些数据结构，如HashMap 和ArrayList。

当我尝试遍历驱动程序中的 RDD 对象并尝试获取 hashmap 元素时，它会抛出 Serialization exception。

java.io.IOException: com.esotericsoftware.kryo.KryoException: java.lang.NullPointerException

示例程序：

驱动类：

主要实现：

public class MyMainTest {
    public static ArrayList<MyInterface> call() throws Exception{
        ArrayList<MyInterface> li = new ArrayList<MyInterface>();
        for(int i=0;i<3;i++){
            MyRecord  myrec = new MyRecord();
            li.add(myrec.addToTagMap());
        }
        return li;
    }
}

Main Implementation:
public class MyMainTest {
    public static ArrayList<MyInterface> call() throws Exception{
        ArrayList<MyInterface> li = new ArrayList<MyInterface>();
        for(int i=0;i<3;i++){
            MyRecord  myrec = new MyRecord();
            li.add(myrec.addToTagMap());
        }
        return li;
    }
}

My Pojo Class:
public class MyRecord implements MyInterface,Serializable{


    protected HashMap<String, Object> allTagNameValueMap_
    =  new HashMap<String, Object>(){
        {
            put("key1", "T1");
            put("key2", "Val2");
            put("key3", 3);
        }};
        protected MyRecord addToTagMap() throws Exception  {
            MyRecord myRec = new MyRecord();
            myRec.allTagNameValueMap_.put("key1", "New Value");
            myRec.allTagNameValueMap_.put("key2", "New Value2");
            myRec.allTagNameValueMap_.put("key4", 22);

            return  myRec;
        }
        @Override
        public Object getKey1() {
            // TODO Auto-generated method stub
            return allTagNameValueMap_.get("key1");
        }
        @Override
        public Object getKey2() {
            // TODO Auto-generated method stub
            return allTagNameValueMap_.get("key2");
        }
        @Override
        public Object getKey3() {
            // TODO Auto-generated method stub
            return allTagNameValueMap_.get("key3");
        }
}

My Interface:
public interface MyInterface extends Serializable{

    public Object getKey1();
    public Object getKey2();
    public Object getKey3();
}

序列化跟踪：

allTagNameValueMap_ (com.org.util.ex.ExRecord)

at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1140)
at org.apache.spark.rdd.ParallelCollectionPartition.readObject(ParallelCollectionRDD.scala:70)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:606)
at java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1017)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1893)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1990)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1915)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:370)
at org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:68)
at org.apache.spark.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:94)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:185)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:745)

原因：com.esotericsoftware.kryo.KryoException: java.lang.NullPointerException 序列化跟踪：

allTagNameValueMap_ (com.org.util.ex.ExRecord)

at com.esotericsoftware.kryo.serializers.FieldSerializer$ObjectField.read(FieldSerializer.java:626)
at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:221)
at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:729)
at com.esotericsoftware.kryo.serializers.DefaultArraySerializers$ObjectArraySerializer.read(DefaultArraySerializers.java:338)
at com.esotericsoftware.kryo.serializers.DefaultArraySerializers$ObjectArraySerializer.read(DefaultArraySerializers.java:293)
at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:729)
at com.twitter.chill.WrappedArraySerializer.read(WrappedArraySerializer.scala:36)
at com.twitter.chill.WrappedArraySerializer.read(WrappedArraySerializer.scala:23)
at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:729)
at org.apache.spark.serializer.KryoDeserializationStream.readObject(KryoSerializer.scala:138)
at org.apache.spark.rdd.ParallelCollectionPartition$$anonfun$readObject$1$$anonfun$apply$mcV$sp$2.apply(ParallelCollectionRDD.scala:80)
at org.apache.spark.rdd.ParallelCollectionPartition$$anonfun$readObject$1$$anonfun$apply$mcV$sp$2.apply(ParallelCollectionRDD.scala:80)
at org.apache.spark.util.Utils$.deserializeViaNestedStream(Utils.scala:130)
at org.apache.spark.rdd.ParallelCollectionPartition$$anonfun$readObject$1.apply$mcV$sp(ParallelCollectionRDD.scala:80)
at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1137)
... 20 more

原因：java.lang.NullPointerException 在 java.util.HashMap.put(HashMap.java:493) 在 com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:135) 在 com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:17) 在 com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:648) 在 com.esotericsoftware.kryo.serializers.FieldSerializer$ObjectField.read(FieldSerializer.java:605)

在Spark中序列化HashMap时有什么特别要注意的吗？

【问题讨论】：

能否请您编辑一下日志，使其更易于阅读
欢迎来到 SO Hari。你能提供一个MCVE吗？它会更容易帮助你，对更多的读者更有用。同时检查this。
谢谢。 zero323 我应该从我的驱动程序调用 MyKryoRegistrator，然后将 sparkConf 可序列化设置为 KryoSerializable
我不想听起来很粗鲁，但我真的建议你试试 Scala，你会看到你的代码看起来更短，更容易理解！我不是说你的代码不好或任何东西！以防万一
是的，Scala 真的很棒。但是我有一些限制，现在必须坚持使用 Java。

标签： java serialization apache-spark rdd

【解决方案1】：

我解决了这个问题。

我遵循的步骤。

创建一个 Kryo 注册器类并将 Pojo 注册为 Kryo Serializable 。

public class MyKryoRegistrator implements KryoRegistrator, Serializable {
  @Override
  public void registerClasses(Kryo kryo) {
      // Product POJO associated to a product Row from the DataFrame      
      kryo.register(MyRecord.class);  
  }
}

2.在我的main里面的spark conf中设置可序列化类型和Kryo注册器类。

sConf.set("spark.serializer","org.apache.spark.serializer.KryoSerializer");
sConf.set("spark.kryo.registrator","com.spark.util.umf.MyKryoRegistrator");

3.使用POJO类中的构造函数初始化HashMap。

受保护的地图 allTagNameValueMap_ =new 哈希映射（）；公共我的记录（）{ allTagNameValueMap_.put("key1", "V1"); allTagNameValueMap_.put("key2", "V2"); allTagNameValueMap_.put("key3", "V3"); };

【讨论】：