我不完全确定这适用于Scala，但在Java中，我通过重构我的代码解决了NotSerializableException，这样闭包就不会访问不可序列化的final字段。

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)

我也遇到过类似的问题，从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(_))
}

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

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引入了

我用另一种方法解决了这个问题。您只需要在传递闭包之前序列化对象，然后反序列化。即使您的类不是Serializable，这种方法也很有效，因为它在幕后使用了Kryo。你只需要一些咖喱。；）

下面是我的一个例子:

def genMapper(kryoWrapper: KryoSerializationWrapper[(Foo => Bar)])
               (foo: Foo) : Bar = {
    kryoWrapper.value.apply(foo)
}
val mapper = genMapper(KryoSerializationWrapper(new Blah(abc))) _
rdd.flatMap(mapper).collectAsMap()

object Blah(abc: ABC) extends (Foo => Bar) {
    def apply(foo: Foo) : Bar = { //This is the real function }
}

你可以随心所欲地让Blah变得复杂，比如类、伴生对象、嵌套类、对多个第三方库的引用。

KryoSerializationWrapper是指:https://github.com/amplab/shark/blob/master/src/main/scala/shark/execution/serialization/KryoSerializationWrapper.scala

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