scala - 为什么在 RDD 中，map 会给出 NotSerializedException，而 foreach 不会？

Question

我理解 map 和 foreach (懒惰和渴望)之间的基本区别，我也理解为什么这个代码片段

sc.makeRDD(Seq("a", "b")).map(s => new java.io.ByteArrayInputStream(s.getBytes)).collect

应该给

java.io.NotSerializableException: java.io.ByteArrayInputStream

然后我认为下面的代码片段也应该如此

sc.makeRDD(Seq("a", "b")).foreach(s => {
  val is = new java.io.ByteArrayInputStream(s.getBytes)
  println("is = " + is)
})

但是这段代码运行良好。为什么会这样？

Answer 1

实际上，map 和 foreach 之间的根本区别不是评估策略。让我们看一下签名(为了简洁起见，我省略了 map 的隐式部分):

def map[U](f: (T) ⇒ U): RDD[U]
def foreach(f: (T) ⇒ Unit): Unit 

map 接受从 T 到 U 的函数，将其应用于现有 RDD[T] 的每个元素> 并返回RDD[U]。为了允许像洗牌这样的操作，U 必须是可序列化的。

foreach 接受从 T 到 Unit 的函数(类似于 Java void)并单独使用什么也不返回。一切都发生在本地，不涉及网络流量，因此不需要序列化。与map不同，当想要获得某种副作用时应该使用foreach，例如your previous question .

顺便说一句，这两者实际上是不同的。您在 map 中使用的匿名函数是一个函数:

(s: String) => java.io.ByteArrayInputStream

以及您在 foreach 中使用的一个，如下所示:

(s: String) => Unit

如果您将第二个函数与 map 一起使用，您的代码将编译，尽管结果将与您想要的相距甚远(RDD[Unit])。