本文全部手写原创,请勿复制粘贴、转载请注明出处,谢谢配合!
本节是对前面所讲的 Transformation和Action操作的一个补充,以深入了解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缓存下来,可以存储在内存或者磁盘。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)的概念
窄依赖指的是每一个父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 之间做 shuffle,Stage 之内做 pipeline(流水线)。方便stage内优化。
相比宽依赖,窄依赖对优化很有利:
- 数据的容错性:
假如某个节点出故障了:
窄依赖:只要重算和子RDD分区对应的父RDD分区即可;
宽依赖:极端情况下,所有的父RDD分区都要进行重新计算。
如下图所示,b1分区丢失,则需要重新计算a1,a2和a3,这就产生了冗余计算(a1,a2,a3中对应b2的数据)。
- 传输
宽依赖往往对应着shuffle操作,需要在运行过程中将同一个父RDD的分区传入到不同的子RDD分区中,中间可能涉及多个节点之间的数据传输;而窄依赖的每个父RDD的分区只会传入到一个子RDD分区中,通常可以在一个节点内完成转换。