本文全部手写原创，请勿复制粘贴、转载请注明出处，谢谢配合！

本节是对前面所讲的 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）的概念

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分区中，通常可以在一个节点内完成转换。