I have a piece of Spark code that worked on Spark 1.3 but fails when I move it to Spark 1.5.2 (cluster upgrade out of my control). The failure is as follows:
Caused by: java.io.NotSerializableException: com.location.model.Profile
Serialization stack:
- object not serializable (class: com.location.model.Profile, value: com.location.model.Profile@596032b0)
- field (class: org.apache.spark.rdd.PairRDDFunctions$$anonfun$aggregateByKey$1, name: zeroValue$3, type: class java.lang.Object)
- object (class org.apache.spark.rdd.PairRDDFunctions$$anonfun$aggregateByKey$1, <function0>)
- field (class: org.apache.spark.rdd.PairRDDFunctions$$anonfun$aggregateByKey$1$$anonfun$1, name: $outer, type: class org.apache.spark.rdd.PairRDDFunctions$$anonfun$aggregateByKey$1)
- object (class org.apache.spark.rdd.PairRDDFunctions$$anonfun$aggregateByKey$1$$anonfun$1, <function0>)
- field (class: org.apache.spark.rdd.PairRDDFunctions$$anonfun$aggregateByKey$1$$anonfun$apply$10, name: createZero$1, type: interface scala.Function0)
- object (class org.apache.spark.rdd.PairRDDFunctions$$anonfun$aggregateByKey$1$$anonfun$apply$10, <function1>)
at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)
at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:47)
at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:84)
at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:301)
The interesting piece is that the class at hand, Profile, is declared as class Profile() extends KryoSerializable and overrides the read/write methods for that interface.
I've also set this config to Spark-submit: "--conf" -> "'spark.serializer=org.apache.spark.serializer.KryoSerializer'" and registered the Profile class with Kryo by doing conf.registerKryoClasses(Array(
classOf[Profile], ...
So all according to instruction in the Spark Tunning guide, and it worked nicely before.
Note that the exception shows a JavaSerializerInstance being used by the ClosureCleaner, and indeed if I add extends Serializable to the Profile class it works. But I'm not sure why it is using that serializer nor why should I be compatible with Java Serialization if I'm specifically asking for Kryo.
Edit: I even removed the parameter altogether, since the code under registerKryoClasses sets the property in any case. In fact, I suspect Kryo serialization is being used (I added a println inside write and it appears, but some kind of previous validation is incorrect).