Spark + Scala转换,不变性和内存消耗开销

Question

我在Youtube上看过一些关于Spark架构的视频.

即使Lazy评估,在出现故障时数据创建的弹性,良好的函数式编程概念是Resilenace分布式数据集成功的原因,但令人担忧的一个因素是由于多次转换导致的内存开销导致数据不可变导致的内存开销.

如果我正确地理解了这个概念,那么每个转换都会创建新的数据集,因此内存需求将会多次消失.如果我在代码中使用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 and reduceByKey.这是否意味着对于X大小的数据我需要3X数据内存？

我的理解是否正确？缓存RDD只是解决此问题的解决方案吗？

一旦我开始缓存,它可能会溢出到磁盘,因为大尺寸和性能会因磁盘IO操作而受到影响.在这种情况下,Hadoop和Spark的性能是否相当？

编辑:

从答案和评论中,我已经理解了延迟初始化和管道流程.我假设3 X内存,其中X是初始RDD大小不准确.

但是有可能在内存中缓存1 X RDD并通过pipleline更新它吗？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,以便我们可以看到这对单个工作者的影响.然后我println为每个转换添加一个,以便我们可以看到工作流程如何移动.您看到它处理该行,然后它处理该行的输出,然后进行缩减.因此,如您所建议的那样,每次转换都没有存储单独的状态.相反,每个数据都会在整个转换过程中循环,直到需要进行随机播放,这可以从UI的DAG可视化中看出:

这是懒惰的胜利.至于Spark v Hadoop,已经有很多东西(只是谷歌),但要点是Spark倾向于利用开箱即用的网络带宽,在那里给它一个提升.然后,通过懒惰获得了许多性能改进,特别是如果已知模式并且您可以使用DataFrames API.

因此,总体而言,Spark几乎可以在任何方面击败MR.