地图不能在scala中序列化？

Question

我是 Scala 新手。为什么“地图”功能不可序列化？如何使其可序列化？例如，如果我的代码如下：

val data = sc.parallelize(List(1,4,3,5,2,3,5))

def myfunc(iter: Iterator[Int]) : Iterator[Int] = {
  val lst = List(("a", 1),("b", 2),("c",3), ("a",2))
  var res = List[Int]()
  while (iter.hasNext) {
    val cur = iter.next
    val a = lst.groupBy(x => x._1).mapValues(_.size)
    //val b= a.map(x => x._2)
    res = res ::: List(cur)
  }
  res.iterator
}

data.mapPartitions(myfunc).collect

如果我取消注释该行

val b= a.map(x => x._2)

代码返回异常：

org.apache.spark.SparkException: Task not serializable
Caused by: java.io.NotSerializableException: scala.collection.immutable.MapLike$$anon$2
Serialization stack:
    - object not serializable (class: scala.collection.immutable.MapLike$$anon$2, value: Map(1 -> 3))
    - field (class: $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC, name: a, type: interface scala.collection.immutable.Map)

非常感谢。

Answer 1

众所周知的 scala 错误：https://issues.scala-lang.org/browse/SI-7005 Map#mapValues is not serializable

我们的 Spark 应用程序中有这个问题，map(identity) 解决了这个问题

rdd.groupBy(_.segment).mapValues(v => ...).map(identity)

Answer 2

下面提供了mapValues 函数的实际实现，正如您所见，它不可序列化并且仅创建一个视图，而不是正确存在的数据，因此您会收到此错误。情境方面的mapValues 可以有很多优势。

protected class MappedValues[C](f: B => C) extends AbstractMap[A, C] with DefaultMap[A, C] {
    override def foreach[D](g: ((A, C)) => D): Unit = for ((k, v) <- self) g((k, f(v)))
    def iterator = for ((k, v) <- self.iterator) yield (k, f(v))
    override def size = self.size
    override def contains(key: A) = self.contains(key)
    def get(key: A) = self.get(key).map(f)
}

Answer 3

您是否尝试过在应用程序中运行相同的代码？我怀疑这是火花外壳的问题。如果您想让它在 spark shell 中工作，那么您可以尝试将 myfunc 的定义及其应用程序包装在花括号中，如下所示：

val data = sc.parallelize(List(1,4,3,5,2,3,5))

val result = { 
  def myfunc(iter: Iterator[Int]) : Iterator[Int] = {
    val lst = List(("a", 1),("b", 2),("c",3), ("a",2))
    var res = List[Int]()
    while (iter.hasNext) {
      val cur = iter.next
      val a = lst.groupBy(x => x._1).mapValues(_.size)
      val b= a.map(x => x._2)
      res = res ::: List(cur)
    }
    res.iterator
  }
  data.mapPartitions(myfunc).collect
}