SparkRDD宽窄依赖及Stage划分

1.术语解释：

Master(Standalone):资源管理的主节点（进程）

Cluster Manager：在集群上获取资源的外部服务（例如standalone，Mesos，Yarn）

Worker Node(standalone):资源管理的从节点（进程）或者说管理本机资源的进程

Application：基于Spark的用户程序，包含了Driver程序和运行在集群上的executor程序

Driver Program：用来连接工作进程（Worker）的程序

Executor：是在一个Worker进程所管理的节点上为某一个Application启动的一个进程，该进程负责运行任务，并且负责将数据存在内存或者磁盘上，每个应用各自独立的executors

Task：被送到某个executor上的工作单元

Job：包含很多任务（Task）的并行计算，可以看做和action对应

Stage：一个Job会被拆分成很多组任务，每组任务被称为Stage

按照资源层面划分：Master ->Worker->Executor->ThreadPool

按照任务层面划分：Application->job->stage->tasks

2.宽窄依赖：

RDD之间有一系列的依赖关系，依赖关系又分为窄依赖和宽依赖。

Spark中的Stage其实是一组并行的任务，任务是一个个的Task

窄依赖：

父RDD和子RDDpartition之间的关系是一对一的，或者父RDD一个partition只对应一个子RDD的partition情况下的父RDD和子RDD partition关系是多对一的，不会有shuffle产生。父RDD的一个分区去到了子RDD的一个分区

宽依赖：

父RDD与子RDD partition之间的关系是一对多，会有shuffle的产生。父RDD的一个分区的数据去到了子RDD的不同分区里面。

区分宽窄依赖主要就是看父RDD的一个partition的流向，要是流向一个的话就是窄依赖，流向多个的话就是宽依赖。相比于宽依赖，窄依赖对优化很有利，主要基于一下两点：

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

2.当RDD分区丢失时（某个节点故障），spark会对数据进行重算

1).对于窄依赖，由于父RDD的一个分区只对应一个子RDD分区，这样只需要重算和子RDD分区对应的父RDD分区即可，所以这个重算对数据的利用率是100%的。

2).对于宽依赖，重算的父RDD分区对应多个字RDD分区，这样实际上父RDD中只有一部分的数据是被用于恢复这个丢失的子RDD分区的，另一部分对应子RDD的其他未丢失分区，这就造成了多余的计算，宽依赖中子RDD分区通常来自于多个父RDD分区，极端情况下，所有的父RDD分区都要重新计算

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

区分这两种依赖很有用，首先，窄依赖允许在一个集群节点上以流水线的方式（pipeline）计算所有父分区。例如，逐个元素地执行map，然后filter操作；而宽依赖则需要首先计算好所有父分区数据，然后在节点间进行shuffle，这和MapReduce类似。第二，窄依赖能够更有效地进行失效节点的恢复，即只需要重新计算丢失RDD分区的父分区，而且不同节点间可以并行计算；而对于一个宽依赖关系的Lineage图，单个节点失效可能导致这个RDD的所有祖先丢失部分分区，因而需要整体重新计算。

在深入分区级别来看待这个问题，重算的效用并不在于算了多少，而是在于有多少是冗余的计算。窄依赖中需要重算的都是必须的，所以重算并不会产生冗余计算。

3.Stage划分：

Spark任务会根据RDD之间的依赖关系，形成一个DAG有向无环图，DAG会提交给DAGScheduler，DAGScheduler会把DAG划分成互相依赖的多个stage，划分stage的依据就是RDD之间的宽窄依赖。遇到宽依赖就划分stage，每个stage包含一个或多个task任务，然后将这些task以taskSet的形式提交给TaskScheduler运行。

stage是由一组并行的task组成。

stage切割规则：从后往前，遇到宽依赖就切割stage

stage计算模式：pipeline管道计算模式，pipeline只是一种计算思想，模式

备注：

1.Spark的pipeline的计算模式：

相当于执行了一个高阶函数f4(f3(f2(f1(&＃8220;&＃8230;..))))。也就是来一条数据然后计算一条数据，把所有的逻辑走完，然后落地。准确的说，是一个task处理一串分区的数据。整个计算逻辑全部走完。而MapReduce是1+1=2,2+1=3的模式，也就是计算完落地，然后拉取，再执行计算，然后再落地到磁盘或者内存，最后数据是落在计算节点上，按reduce的hash分区落地。所以这也是Saprk比MapReduce快的原因，是完全基于内存计算的。

2.管道中的数据何时落地：

shuffle write的时候

对RDD进行持久化的时候

3.Stage的task并行度是由stage的最后一个RDD的分区数来决定的。

一般来说，一个partition对应一个task，但最后reduce的时候，可以手动改变reduce的个数来提高并行度，也就是分区数。例如reduceByKey(xxx,3),GroupByKey(4),union的分区数，是由前面的相加

测试验证pipeline计算模式（迭代器模式）：

import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object pipeLineTest { def main(args: Array[String]): Unit = { val cOnf= new SparkConf().setMaster(&＃8220;local&＃8221;).setAppName(&＃8220;pipelineTest&＃8221;) val sc = new SparkContext(conf) sc.setLogLevel(&＃8220;Error&＃8221;) val rdd1 = sc.parallelize(Array(1,2,3,4)) val rdd2: RDD[Int] = rdd1.map { x => { println(&＃8220;map&＃8212;&＃8212;&＃8212;&＃8221; + x) x } } val rdd3: RDD[Int] = rdd2.filter(x => { println(&＃8220;filter***********&＃8221; + x) true }) rdd3.collect() sc.stop() } }

运行结果如下所示：