小开

最佳答案

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

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

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

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

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

在第二种情况下发生的情况是，您正在调用一个方法，该方法在map函数内部的testing类中定义。Spark看到了这一点，因为方法不能单独序列化，所以Spark尝试序列化整个 testing类，这样代码在另一个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
}

类似的，但不相同的类序列化问题可以让你感兴趣，你可以阅读在2013年的Spark峰会上。

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

编辑(2015-03-15):火星- 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>)

小开

我用另一种方法解决了这个问题。您只需要在传递闭包之前序列化对象，然后反序列化。即使您的类不是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

小开

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

里写道提供了一个很好的解决方案，但解决方案可以更简洁和通用:

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

小开

完整的演讲充分解释了这个问题，并提出了一个很好的范式转换方法来避免这些序列化问题:https://github.com/samthebest/dump/blob/master/sams-scala-tutorial/serialization-exceptions-and-memory-leaks-no-ws.md

投票最多的答案基本上是建议抛弃整个语言特性——不再使用方法，只使用函数。的确，在函数式编程中，类中的方法应该避免，但是将它们转换为函数并不能解决设计问题(参见上面的链接)。

在这种特殊情况下，你可以使用@transient注释告诉它不要试图序列化有问题的值(这里，Spark.ctx是一个自定义类，而不是Spark的OP命名之后的一个类):

@transient
val rddList = Spark.ctx.parallelize(list)

您还可以重新构造代码，使rddList存在于其他地方，但这也很麻烦。

未来可能是孢子

在将来，Scala将包括这些被称为“孢子”的东西，这将允许我们细粒度地控制闭包到底会吸引什么，不吸引什么。此外，这应该将所有意外拉入不可序列化类型(或任何不需要的值)的错误转变为编译错误，而不是现在可怕的运行时异常/内存泄漏。

http://docs.scala-lang.org/sips/pending/spores.html

关于Kryo序列化的提示

当使用kyro，使注册是必要的，这将意味着你得到错误，而不是内存泄漏:

最后，我知道kryo有kryo. setregistrationoptional (true)，但我很难弄清楚如何使用它。当打开这个选项时，如果我没有注册类，kryo似乎仍然会抛出异常。”

向kryo注册类的策略

当然，这只能提供类型级别的控制，而不是值级别的控制。

．.．还会有更多的想法。

小开

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

小开

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

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

小开

在Spark 2.4中，很多人可能会遇到这个问题。Kryo序列化已经变得更好，但在许多情况下，您不能使用spark.kryo.unsafe=true或幼稚的Kryo序列化器。

为了快速修复，请尝试在Spark配置中更改以下内容

spark.kryo.unsafe="false"

或

spark.serializer="org.apache.spark.serializer.JavaSerializer"

通过使用显式广播变量和新的内置twitter-chill api，我修改了我遇到或亲自编写的自定义RDD转换，将它们从rdd.map(row =>转换为rdd.mapPartitions(partition => {函数。

例子

老方法(不太好)

val sampleMap = Map("index1" -> 1234, "index2" -> 2345)
val outputRDD = rdd.map(row => {
val value = sampleMap.get(row._1)
value
})

替代(更好的)方式

import com.twitter.chill.MeatLocker
val sampleMap = Map("index1" -> 1234, "index2" -> 2345)
val brdSerSampleMap = spark.sparkContext.broadcast(MeatLocker(sampleMap))


rdd.mapPartitions(partition => {
val deSerSampleMap = brdSerSampleMap.value.get
partition.map(row => {
val value = sampleMap.get(row._1)
value
}).toIterator
})

这种新方法将每个分区只调用广播变量一次，这更好。如果不注册类，仍然需要使用Java Serialization。

小开

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) < / p >

解决方案-

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)

小开

我也有过类似的经历。

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

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

以前的无效代码

  private final PhoneNumberUtil phoneUtil = PhoneNumberUtil.getInstance();

工作代码

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

学分:

小开

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

小开

Scala类中定义的方法是不可序列化的，方法可以转换为函数来解决序列化问题。

方法的语法

def func_name (x String) : String = {
...
return x
}

函数的语法

val func_name = { (x String) =>
...
x
}

任务不可序列化:java.io.NotSerializableException，当只对类而不是对象调用闭包外部的函数时