scala - Spark + Scala 转换、不变性和内存消耗开销

Question

我在 Youtube 上浏览了一些关于 Spark 的视频架构。

尽管惰性评估、发生故障时数据创建的弹性、良好的函数式编程概念是 Resilenace 分布式数据集成功的原因，但一个令人担忧的因素是由于多个 transformations 造成的内存开销。由于数据不变性导致内存开销。

如果我正确理解这个概念，每次转换都会创建新的数据集，因此内存需求会减少很多次。如果我在我的代码中使用 10 次转换，将创建 10 组数据集，我的内存消耗将增加 10 倍。

例如

val textFile = sc.textFile("hdfs://...")
val counts = textFile.flatMap(line => line.split(" "))
                 .map(word => (word, 1))
                 .reduceByKey(_ + _)
counts.saveAsTextFile("hdfs://...")

以上示例具有三个转换:flatMap、map 和 reduceByKey。这是否意味着 X 大小的数据需要 3 倍的数据内存？

我的理解正确吗？缓存 RDD 是解决这个问题的唯一方法吗？

一旦我开始缓存，它可能会溢出到磁盘，因为它的大小很大，并且性能会因磁盘 IO 操作而受到影响。那么，Hadoop 和 Spark 的性能具有可比性吗？

编辑:

从回答和评论中，我了解了惰性初始化和流水线过程。我对 3 X 内存的假设是不准确的，其中 X 是初始 RDD 大小。

但是否可以在内存中缓存 1 X RDD 并通过管道更新它？ cache() 是如何工作的？

Answer 1

首先，惰性执行意味着可以进行功能组合:

scala> val rdd = sc.makeRDD(List("This is a test", "This is another test", 
                                 "And yet another test"), 1)
rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[70] at makeRDD at <console>:27

scala> val counts = rdd.flatMap(line => {println(line);line.split(" ")}).
     | map(word => {println(word);(word,1)}).
     | reduceByKey((x,y) => {println(s"$x+$y");x+y}).
     | collect
This is a test
This
is
a
test
This is another test
This
1+1
is
1+1
another
test
1+1
And yet another test
And
yet
another
1+1
test
2+1
counts: Array[(String, Int)] = Array((And,1), (is,2), (another,2), (a,1), (This,2), (yet,1), (test,3))

首先请注意，我将并行度强制降低到 1，以便我们可以看到它在单个 worker 上的效果。然后我将 println 添加到每个转换中，以便我们可以看到工作流是如何移动的。您会看到它处理该行，然后处理该行的输出，然后进行归约。因此，没有像您建议的那样为每个转换存储单独的状态。相反，每条数据都在整个转换过程中循环，直到需要洗牌为止，从 UI 的 DAG 可视化可以看出:

那是懒惰的胜利。至于 Spark v Hadoop，那里已经有很多(只需谷歌一下)，但要点是 Spark 倾向于开箱即用地利用网络带宽，从而提高它的性能。然后，通过惰性获得了一些性能改进，尤其是在架构已知并且您可以使用 DataFrames API 的情况下。

因此，总的来说，Spark 在几乎所有方面都轻而易举地击败了 MR。