我也有过类似的经历。

当我在驱动程序(master)上初始化一个变量时触发了错误，但随后试图在其中一个工人上使用它。 当这种情况发生时，Spark Streaming将尝试序列化对象以将其发送给worker，如果对象不可序列化则失败。

我通过使变量为静态来解决这个错误。

以前的无效代码

  private final PhoneNumberUtil phoneUtil = PhoneNumberUtil.getInstance();

工作代码

  private static final PhoneNumberUtil phoneUtil = PhoneNumberUtil.getInstance();

学分:

https://learn.microsoft.com/en-us/answers/questions/35812/sparkexception-job-aborted-due-to-stage-failure-ta.html (pradeepcheekatla-msft的答案) https://databricks.gitbooks.io/databricks-spark-knowledge-base/content/troubleshooting/javaionotserializableexception.html

rdd扩展了Serialisable接口，所以这不是导致任务失败的原因。现在这并不意味着你可以用Spark序列化一个RDD并且避免NotSerializableException

Spark是一个分布式计算引擎，它的主要抽象是一个弹性分布式数据集(RDD)，它可以被视为一个分布式集合。基本上，RDD的元素是在集群的节点上进行分区的，但是Spark将其从用户那里抽象出来，让用户与RDD(集合)交互，就像它是一个本地的集合一样。

不涉及太多细节，但当你在RDD上运行不同的转换(map, flatMap, filter和其他)时，你的转换代码(闭包)是:

在驱动节点上序列化， 发送到集群中的适当节点， 反序列化, 最后在节点上执行

当然，您可以在本地运行(如您的示例)，但所有这些阶段(除了通过网络传输)仍然会发生。[这可以让您在部署到生产环境之前捕捉任何错误]

在第二种情况下，您正在调用一个方法，该方法是在map函数内部的类测试中定义的。Spark看到了这一点，因为方法不能单独序列化，所以Spark尝试序列化整个测试类，这样代码在另一个JVM中执行时仍然可以工作。你有两种可能:

要么你让类测试可序列化，这样整个类都可以被Spark序列化:

import org.apache.spark.{SparkContext,SparkConf}

object Spark {
  val ctx = new SparkContext(new SparkConf().setAppName("test").setMaster("local[*]"))
}

object NOTworking extends App {
  new Test().doIT
}

class Test extends java.io.Serializable {
  val rddList = Spark.ctx.parallelize(List(1,2,3))

  def doIT() =  {
    val after = rddList.map(someFunc)
    after.collect().foreach(println)
  }

  def someFunc(a: Int) = a + 1
}

或者你让someFunc函数而不是方法(函数在Scala中是对象)，这样Spark就可以序列化它:

import org.apache.spark.{SparkContext,SparkConf}

object Spark {
  val ctx = new SparkContext(new SparkConf().setAppName("test").setMaster("local[*]"))
}

object NOTworking extends App {
  new Test().doIT
}

class Test {
  val rddList = Spark.ctx.parallelize(List(1,2,3))

  def doIT() =  {
    val after = rddList.map(someFunc)
    after.collect().foreach(println)
  }

  val someFunc = (a: Int) => a + 1
}

类似的，但不相同的类序列化问题可能会引起您的兴趣，您可以在Spark Summit 2013的演讲中阅读它。

作为旁注，你可以将rddList.map(someFunc(_))重写为rddList.map(someFunc)，它们是完全相同的。通常情况下，首选第二种，因为它不那么冗长，读起来更清楚。

编辑(2015-03-15):Spark -5307引入了SerializationDebugger, Spark 1.3.0是第一个使用它的版本。它将序列化路径添加到NotSerializableException。当遇到NotSerializableException时，调试器访问对象图以查找无法序列化的对象的路径，并构造信息以帮助用户找到该对象。

在OP的例子中，这是打印到stdout的内容:

Serialization stack:
    - object not serializable (class: testing, value: testing@2dfe2f00)
    - field (class: testing$$anonfun$1, name: $outer, type: class testing)
    - object (class testing$$anonfun$1, <function1>)

Grega的回答很好地解释了为什么原始代码不能工作以及解决这个问题的两种方法。然而，这种解决方案不是很灵活;考虑这样一种情况，闭包包含一个对你无法控制的非serializable类的方法调用。您既不能向该类添加Serializable标记，也不能更改底层实现以将方法更改为函数。

Nilesh提出了一个很好的解决方案，但解决方案可以更简洁和通用:

def genMapper[A, B](f: A => B): A => B = {
  val locker = com.twitter.chill.MeatLocker(f)
  x => locker.get.apply(x)
}

这个函数序列化器可以用来自动封装闭包和方法调用:

rdd map genMapper(someFunc)

这种技术还有一个好处，就是不需要额外的Shark依赖来访问KryoSerializationWrapper，因为Twitter的Chill已经被核心Spark引入了

我也遇到过类似的问题，从Grega的回答中我理解到

object NOTworking extends App {
 new testing().doIT
}
//adding extends Serializable wont help
class testing {

val list = List(1,2,3)

val rddList = Spark.ctx.parallelize(list)

def doIT =  {
  //again calling the fucntion someFunc 
  val after = rddList.map(someFunc(_))
  //this will crash (spark lazy)
  after.collect().map(println(_))
}

def someFunc(a:Int) = a+1

}

你的doIT方法试图序列化someFunc(_)方法，但由于方法是不可序列化的，它试图序列化类测试，这也是不可序列化的。

为了让你的代码工作，你应该在doIT方法中定义someFunc。例如:

def doIT =  {
 def someFunc(a:Int) = a+1
  //function definition
 }
 val after = rddList.map(someFunc(_))
 after.collect().map(println(_))
}

如果有多个函数进入图中，那么所有这些函数都应该对父上下文可用。

我的解决方案是添加一个同伴类来处理类中不可序列化的所有方法。

def upper(name: String) : String = { 
var uppper : String  =  name.toUpperCase()
uppper
}

val toUpperName = udf {(EmpName: String) => upper(EmpName)}
val emp_details = """[{"id": "1","name": "James Butt","country": "USA"},
{"id": "2", "name": "Josephine Darakjy","country": "USA"},
{"id": "3", "name": "Art Venere","country": "USA"},
{"id": "4", "name": "Lenna Paprocki","country": "USA"},
{"id": "5", "name": "Donette Foller","country": "USA"},
{"id": "6", "name": "Leota Dilliard","country": "USA"}]"""

val df_emp = spark.read.json(Seq(emp_details).toDS())
val df_name=df_emp.select($"id",$"name")
val df_upperName= df_name.withColumn("name",toUpperName($"name")).filter("id='5'")
display(df_upperName)

这会产生错误 sparkexception:任务不可序列化 org.apache.spark.util.ClosureCleaner .ensureSerializable美元(ClosureCleaner.scala: 304)

解决方案-

import java.io.Serializable;

object obj_upper extends Serializable { 
  def upper(name: String) : String = 
  {
    var uppper : String  =  name.toUpperCase()
    uppper
  }
val toUpperName = udf {(EmpName: String) => upper(EmpName)}
}

val df_upperName= 
df_name.withColumn("name",obj_upper.toUpperName($"name")).filter("id='5'")
display(df_upperName)