HashMap的火花序列化问题

我想序列化一个Java类，其对象将在Java Spark RDD中收集。 POJO包含一些数据结构，如HashMap和ArrayList。HashMap的火花序列化问题

当我试图迭代我的驱动程序中的RDD对象并尝试获取hashmap元素时，它抛出了Serialization exception。

产生java.io.IOException：com.esotericsoftware.kryo.KryoException：显示java.lang.NullPointerException

示例程序：

驱动程序类：

主要Implementaion：

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(); 
}

序列化轨迹：

allTagNam eValueMap_（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.util.HashMap.put（HashMap.java:493显示java.lang.NullPointerException ）在com.esotericsoftware.kryo.serializers.MapSerializer.read（ MapSerializer.java:135 ） at com.esotericsoftware.kryo.serializers.MapSerializer.read（MapSerializer.java:17） at com.esotericsoftware.kryo.Kryo.readObject（Kryo.java:648） at com.esotericsoftware.kryo.serializers.FieldSerializer $ ObjectField.read（FieldSerializer.java:605）

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

来源

2015-10-15 Harisyam

请问您可以编辑日志使其更容易阅读 –

欢迎来到SO Hari。您可以提供[MCVE]（http://stackoverflow.com/help/mcve）吗？这对帮助你更容易，对更多的读者更有用。在此期间检查[this]（http://stackoverflow.com/a/33042316/1560062）。 – zero323

谢谢。 zero323 我应该对MyKryoRegistrator一个电话从我的驱动程序的，然后设置sparkConf序列化作为KryoSerializable – Harisyam

我修正了这个问题。

我遵循的步骤。

创建一个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.将序列化类型，并在里面我的主要火花conf下KRYO registrator类。

sConf.set（“spark.serializer”，“org.apache.spark.serializer.KryoSerializer”）;
sConf.set（“spark.kryo.registrator”，“com.spark.util.umf.MyKryoRegistrator”）;

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

受保护的地图allTagNameValueMap_ = new HashMap（）; public MyRecord（）{ allTagNameValueMap_.put（“key1”，“V1”）; allTagNameValueMap_.put（“key2”，“V2”）; allTagNameValueMap_.put（“key3”，“V3”）; };

来源

2015-10-22 00:04:01 Harisyam

HashMap的火花序列化问题

回答

相关问题