当类扩展可序列化时，Apache Spark 任务不可序列化答案

【问题标题】：Apache Spark Task not Serializable when Class exends Serializable当类扩展可序列化时，Apache Spark 任务不可序列化
【发布时间】：2015-06-17 21:33:14
【问题描述】：

我一直有关于 Task not Serializable 的错误。

我创建了一个小类，它扩展了 Serializable - 我认为当您需要对其中的值进行序列化时就是这种情况。

class SGD(filePath : String) extends Serializable {

 val rdd = sc.textFile(filePath)
 val mappedRDD = rdd.map(x => x.split(" ")
                    .slice(0,3))
                    .map(y => Rating(y(0).toInt, y(1).toInt, y(2).toDouble))
                    .cache

 val RNG = new Random(1)
 val factorsRDD = mappedRDD(x => (x.user, (x.product, x.rating)))
                 .groupByKey
                 .mapValues(listOfItemsAndRatings => 
                                             Vector(Array.fill(2){RNG.nextDouble}))
}

最后一行总是导致Task not Serializable 错误。我不明白的是：类是可序列化的；并且，Random 类也可以根据API 进行序列化。那么，我做错了什么？我一直无法让这样的东西发挥作用；因此，我想我的理解是错误的。我一直被告知该类必须是可序列化的......好吧，它仍然不起作用！？

【问题讨论】：

Rating 可以序列化吗？
我可以理解您遵循 scala 约定在一行中编写所有功能。但是请尝试拆分这一行并分别编写每个功能，以便您可以找出瓶颈。

标签： scala serialization apache-spark rdd

【解决方案1】：

scala.util.Random 直到2.11.0-M2 才可序列化。

您很可能使用的是早期版本的 Scala。

一个类在其所有成员都可序列化之前不会变为可序列化（或提供一些其他机制来序列化它们，例如transient 或readObject/writeObject。）

在 spark-1.3 中运行给定示例时，我得到以下堆栈跟踪：

Caused by: java.io.NotSerializableException: scala.util.Random
Serialization stack:
    - object not serializable (class: scala.util.Random, value: scala.util.Random@52bbf03d)
    - field (class: $iwC$$iwC$SGD, name: RNG, type: class scala.util.Random)

修复它的一种方法是在 mapValues 中设置随机变量：

mapValues(listOfItemsAndRatings => { val RNG = new Random(1)
                   Vector(Array.fill(2)(RNG.nextDouble)) })

【讨论】：

啊哈，如果我记得我使用的是 2.10.4。这样可行;但是，现在每个向量都填充了相同的数字。我需要它，所以每个向量都是不同的数字，或者至少，它们中的大多数是不同的。
@monster 如果是这样，只需在填充方法中使用Random.nextDouble。它采用随机种子值，因此应该产生不同的数字。
遗憾的是，我需要我的测试是可重复的。我找到了解决方法（将mapValues 更改为map，然后使用userid 设置种子。您的回答非常有帮助，非常感谢！
我刚刚使用了java.util.Random，因为我被困在可序列化的 Scala 2.10.6 上。