1. 首页
  2. 大数据
  3. Spark教程

【Spark教程】(六)RDD的惰性求值、持久化及依赖关系

惰性求值


RDD的Transformation操作都是惰性求值的。什么意思呢?也就是说Transformation操作不会开始真正的计算,只有在执行Action操作的时候Spark才会真正开始计算。转化操作不会立刻执行,而是在内部记录下所要执行的操作的相关信息,必要时再执行。

那么这么设计的优势是什么?

比如,我们要筛选出某个文件中包含“spark”的行,最后只输出第一个符合条件的行。

我们先读取文件textFile()生成Rdd1,然后使用filter()方法生成Rdd2,最后是行动操作first()。

试想,如果读取文件的时候就把所有的行都存储起来,但我们马上就要筛选出只具有“Spark”的行,等筛选出来具有“spark”的行,又要求只输出第一个。这样是不是太浪费存储空间了呢?所以实际上,Spark是在行动操作first()的时候开始真正的运算:只扫描第一个匹配的行,不需要读取整个文件。

持久化


  • 为什么需要持久化?
    每当我们调用一个新的Action操作时,整个RDD都会从头开始运算(DAG重算)。如果某个rdd会被反复重用,那么这将非常低效,我们应该对多次使用的RDD进行一个持久化操作。

  • 如何持久化?
    RDD中persist()方法让Spark把这个RDD缓存下来,可以存储在内存或者磁盘,如果是MEMORY_ONLY方式可以直接使用cache()。unpersist()可以将持久化的RDD移除。

scala> val rdd = sc.parallelize(1 to 5)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[5] at parallelize at <console>:24

scala> val rdd1 = rdd.map(_+5)
rdd1: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[6] at map at <console>:26

scala> rdd1 = rdd1.persist(StorageLevel.MEMORY_ONLY)
res2: rdd1.type = MapPartitionsRDD[6] at map at <console>:26

scala> rdd1.reduce(_+_)
res3: Int = 40

scala> rdd1.count()
res4: Long = 5

scala> rdd1.first()
res5: Int = 6

scala> rdd1.unpersist()
res7: rdd1.type = MapPartitionsRDD[6] at map at <console>:26

本例多次对rdd1进行Action操作,如果不使用持久化,每次Action整个RDD都会从头开始运算,使用持久化大大提高效率。

  • 存储级别
    存储级别

依赖关系


前面我们讲过,RDD的Transformation操作是一个RDD生成另一个新的RDD的过程,那么新的RDD必然依赖原来的RDD。那么RDD的依赖分为几种,分别是什么含义呢?为什么要分类呢?作用是什么?

先思考一个问题:下例中(1)和(2),后个RDD在对前一个RDD的依赖上有什么区别呢?

scala> val rdd = sc.parallelize(List(("LYL",1),("LYL",2),("CCC",3),("DDD",4)))
rdd: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[0] at parallelize at <console>:24
//例(1):map+filter
scala> val rddn = rdd.map(x=>(x._1,x._2+3)).filter(x=>(x._2<7))
rddn: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[2] at filter at <console>:26

scala> rddn.collect()
res0: Array[(String, Int)] = Array((LYL,4), (LYL,5), (CCC,6))
//例(2):map+reduceByKey
scala> val rddw = rdd.map(x=>(x._1,x._2+3)).reduceByKey(_+_)
rddw: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[4] at reduceByKey at <console>:26

scala> rddw.collect()
res1: Array[(String, Int)] = Array((CCC,6), (LYL,9), (DDD,7))

例(1)filter()并不依赖map()生成的RDD的所有分区,也就是说即使map没有全部完成我也可以开始filter,先单独对(“LYL”,1)map后进行filter,再单独对(“LYL”,2)map后进行filter,最后得到的结果也完全没问题。

但是例(2)中reduceBykey()必须依赖map后生成的所有分区。必须map全部完成我才可以reduceBykey()。因为我们需要对所有key进行一个分类整合,不可能单独对(“LYL”,1)进行一个reduceBykey操作吧?

那么这就是两种不同的依赖关系。(1)是窄依赖,(2)是宽依赖。下面进入正题:

窄依赖和宽依赖(narrow/wide dependency)的概念

narrow/wide dependency

窄依赖指的是每一个父RDD的Partition最多被子RDD的一个Partition使用。1对1或者n对1(简记:独生子女),如map、filter、union

宽依赖指的是多个子RDD的Partition会依赖同一个父RDD的Partition 。n对n(简记:超生),如 groupByKey、reduceByKey、sortByKey

依赖关系的数据流图

数据流图

根据前面我们讲过的架构原理可知:在Spark中,一段程序实际上构造了一个由相互依赖的多个RDD组成的有向无环图(DAG)。spark会根据RDD之间的依赖关系将DAG图划分为不同的阶段Stage

对于窄依赖,由于partition依赖关系的确定性,partition的转换处理就可以在同一个线程里完成,窄依赖就被spark划分到同一个stage中,如图RDD C、D、E、F它们都在Stage2中。

而对于宽依赖,只能等父RDD shuffle处理完成后,下一个stage才能开始接下来的计算,因此宽依赖要单独划分一个Stage,如上图的RDD A。

区分两种依赖的作用

划分 Stage:Stage的划分依据就是根据依赖关系来的,Stage之间做 shuffle,Stage 之内做 pipeline(流水线)。方便stage内优化。

相比宽依赖,窄依赖对优化很有利:

  • 数据的容错性:

假如某个节点出故障了:

窄依赖:只要重算和子RDD分区对应的父RDD分区即可;

宽依赖:极端情况下,所有的父RDD分区都要进行重新计算。

如下图所示,b1分区丢失,则需要重新计算a1,a2和a3,这就产生了冗余计算(a1,a2,a3中对应b2的数据)。

宽依赖的容错性

  • 传输

宽依赖往往对应着shuffle操作,需要在运行过程中将同一个父RDD的分区传入到不同的子RDD分区中,中间可能涉及多个节点之间的数据传输;而窄依赖的每个父RDD的分区只会传入到一个子RDD分区中,通常可以在一个节点内完成转换。

BDStar原创文章。发布者:Liuyanling,转载请注明出处:http://bigdata-star.com/archives/910

发表评论

登录后才能评论

联系我们

562373081

在线咨询:点击这里给我发消息

邮件:562373081@qq.com

工作时间:周一至周五,9:30-18:30,节假日休息

QR code