为什么这个Spark代码抛出java.io.NotSerializableException

Question

我想访问RDD中转换内的伴随对象的方法。为什么下面不工作：

import org.apache.spark.rdd.RDD 
import spark.implicits._ 
import org.apache.spark.sql.{Encoder, Encoders} 

class Abc { 
    def transform(x: RDD[Int]): RDD[Double] = { x.map(Abc.fn) } 
} 

object Abc { 
    def fn(x: Int): Double = { x.toDouble } 
} 

implicit def abcEncoder: Encoder[Abc] = Encoders.kryo[Abc] 

new Abc().transform(sc.parallelize(1 to 10)).collect 

上面这段代码抛出一个java.io.NotSerializableException：

org.apache.spark.SparkException: Task not serializable 
    at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:298) 
    at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:288) 
    at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:108) 
    at org.apache.spark.SparkContext.clean(SparkContext.scala:2094) 
    at org.apache.spark.rdd.RDD$$anonfun$map$1.apply(RDD.scala:370) 
    at org.apache.spark.rdd.RDD$$anonfun$map$1.apply(RDD.scala:369) 
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151) 
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112) 
    at org.apache.spark.rdd.RDD.withScope(RDD.scala:362) 
    at org.apache.spark.rdd.RDD.map(RDD.scala:369) 
    at Abc.transform(<console>:19) 
    ... 47 elided 
Caused by: java.io.NotSerializableException: Abc 
Serialization stack: 
     - object not serializable (class: Abc, value: Abc@4f598dfb) 
     - field (class: Abc$$anonfun$transform$1, name: $outer, type: class Abc) 
     - object (class Abc$$anonfun$transform$1, <function1>) 
    at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40) 
    at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:46) 
    at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:100) 
    at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:295) 
    ... 57 more 

即使定义Encoder为ABC类并不在这里帮助。但更重要的问题是，为什么Abc类的对象的系列化被尝试呢？我的第一个想法是伴侣对象是类的单例对象，所以也许会尝试序列化它。但似乎并不喜欢这样，因为当我从另一个类调用Abc.fn：

class Xyz { 
    def transform(x: RDD[Int]): RDD[Double] = { x.map(Abc.fn) } 
} 

implicit def xyzEncoder: Encoder[Xyz] = Encoders.kryo[Xyz] 

new Xyz().transform(sc.parallelize(1 to 10)).collect 

我得到一个java.io.NotSerializableException: Xyz

Answer 1

这里是一个伟大的文章，讨论“序列化”与“非在Apache的星火序列化对象：

Using Non-Serializable Objects in Apache Spark, Nicola Ferraro

本文提供了一些建议，以：

• 这是怎么回事你的具体情况

• 一些替代品让你的对象并不需要“序列化”

Answer 2

主要抽象OS火花RDD这是跨节点分区集群。所以当我们运行RDD时，它在驱动程序节点中被序列化，并且被分配给其他适当的节点。然后工作节点将其反序列化并执行。

在你的情况下，类ABC不能被序列化并分发给其他工作节点。 您需要序列化序列化ABC类

class Abc with Serializable{ 
    def transform(x: RDD[Int]): RDD[Double] = { x.map(Abc.fn) } 
}