火花对象不可序列化[重复]

Question

有没有关于为什么整个对象B需要序列化的想法？

object A{

  def main(args:Array[String])= {
    val conf = new SparkConf().setAppName("test").setMaster("local")
    val sc = new SparkContext(conf)
    val rdd=sc.makeRDD(Seq(1,2,3,4,5,6,7))
    val b=new B
    b.add(rdd)
  }
}
class B {

  val s="456"
  def add=(rdd:RDD[Int])=>{
    rdd.map(e=>e+" "+s).foreach(println)
  }
}

关于“对象不可序列化”的异常：

Serialization stack:
- object not serializable (class: B, value: B@1fde4f40)
- field (class: B$$anonfun$add$1, name: $outer, type: class B)
- object (class B$$anonfun$add$1, <function1>)
- field (class: B$$anonfun$add$1$$anonfun$apply$1, name: $outer, type: class B$$anonfun$add$1)
- object (class B$$anonfun$add$1$$anonfun$apply$1, <function1>)

Answer 1

请参见链接任务not serializable:java。伊奥。仅在类而非对象上调用闭包外函数时发生NotSerializableException。你的语法是什么

def add=(rdd:RDD[Int])=>{
  rdd.map(e=>e+" "+s).foreach(println)
}

真正的意思是

def addReal(): RDD[Int] => Unit = (rdd: RDD[Int]) => {
  val rddFunc: Int => String = e => e + " " + s
  rdd.map(rddFunc).foreach(println)
}

所以现在rddFunc从其包含对象B的上下文中捕获s，这是通过实际捕获不可序列化的整个对象B来完成的。

因此，除了链接答案中列出的选项外，还应该帮助您将添加到

def add(rdd:RDD[Int]) = {
  val localS = s
  rdd.map(e => e + " " + localS).foreach(println)
}

主要技巧是引入局部变量，强制rddFunc只捕获它，而不是整个B。

旁注：在上面的代码中，我还更改了您的添加的签名

def add():Function1[RDD[Int], Unit]  

你必须这么做

def add(rdd:RDD[Int]):Unit

如果add返回一个立即调用的函数，则没有任何好处。