Spark DAG概述

1、 DAG定义

DAG每一个节点表明啥？表明的一个RDD

这里再次复习RDD的5大特性

• 一组分片（Partition），即数据集的基本组成单位。对于RDD来讲，每一个分片都会被一个计算任务处理，并决定并行计算的粒度。用户能够在建立RDD时指定RDD的分片个数，若是没有指定，那么就会采用默认值。默认值就是程序所分配到的CPU Core的数目。
• 一个计算每一个分区的函数。Spark中RDD的计算是以分片为单位的，每一个RDD都会实现compute函数以达到这个目的。compute函数会对迭代器进行复合，不须要保存每次计算的结果。
• RDD之间的依赖关系。RDD的每次转换都会生成一个新的RDD，因此RDD之间就会造成相似于流水线同样的先后依赖关系。在部分分区数据丢失时，Spark能够经过这个依赖关系从新计算丢失的分区数据，而不是对RDD的全部分区进行从新计算。
• 一个Partitioner，即RDD的分片函数。当前Spark中实现了两种类型的分片函数，一个是基于哈希的HashPartitioner，另一个是基于范围的RangePartitioner。只有对于于key-value的RDD，才会有Partitioner，非key-value的RDD的Parititioner的值是None。Partitioner函数不但决定了RDD自己的分片数量，也决定了parent RDD Shuffle输出时的分片数量。
• 一个列表，存储存取每一个Partition的优先位置（preferred location）。对于一个HDFS文件来讲，这个列表保存的就是每一个Partition所在的块的位置。按照“移动数据不如移动计算”的理念，Spark在进行任务调度的时候，会尽量地将计算任务分配到其所要处理数据块的存储位置。

2.

• 一个RDD生成两个RDD：

        RDD2 = RDD1.filter(xxxxx)

        RDD3 = RDD1.filter(yyyy)

       是从RDD1到RDD2，RDD3这样的过程

• Union是两个RDD合并成一个的过程   

        则是RDD2 RDD3变成RDD4的过程

• filter/map/reduceByKey 应该都是一条直线

        是从RDD4到RDD5这样的过程

        上述都是transformation

        RDD5.collect();  //action

        RDD5.foreach();  //action

    这种则会生成两个job，会顺序提交，前一个job执行结束以后才会提交下一个job（假设上述代码都在一个线程中）

RDD依赖关系

RDD依赖关系，也就是有依赖的RDD之间的关系，好比RDD1------->RDD2（RDD1生成RDD2），RDD2依赖于RDD1。这里的生成也就是RDDtransformation操做

窄依赖（也叫narrow依赖）

从父RDD角度看：一个父RDD只被一个子RDD分区使用。父RDD的每一个分区最多只能被一个Child RDD的一个分区使用

从子RDD角度看:依赖上级RDD的部分分区     精确知道依赖的上级RDD分区，会选择和本身在同一节点的上级RDD分区，没有网络IO开销，高效。如map，flatmap，filter

宽依赖（也叫shuffle依赖/wide依赖）

从父RDD角度看：一个父RDD被多个子RDD分区使用。父RDD的每一个分区能够被多个Child RDD分区依赖

从子RDD角度看:依赖上级RDD的全部分区     没法精肯定位依赖的上级RDD分区，至关于依赖全部分区（例如reduceByKey）  计算就涉及到节点间网络传输  

                        父分区，都只有一根箭头                         父分区，都有多个箭头

                        子分区，来自部分父分区                         子分区，来自所有父分区

Spark之因此将依赖分为narrow和 shuffle：

(1) narrow dependencies能够支持在同一个集群Executor上，以pipeline管道形式顺序执行多条命令，例如在执行了map后，紧接着执行filter。分区内的计算收敛，不须要依赖全部分区的数据，能够并行地在不一样节点进行计算。因此它的失败恢复也更有效，由于它只须要从新计算丢失的parent partition便可，

(2)shuffle dependencies 则须要全部的父分区都是可用的，必须等RDD的parent partition数据所有ready以后才能开始计算，可能还须要调用相似MapReduce之类的操做进行跨节点传递。从失败恢复的角度看，shuffle dependencies 牵涉RDD各级的多个parent partition。
 

如图所示，左边的都是右边的父分区

划分stage

因为shuffle依赖必须等RDD的parent RDD partition数据所有ready以后才能开始计算，所以spark的设计是让parent RDD将结果写在本地，彻底写完以后，通知后面的RDD。后面的RDD则首先去读以前的本地数据做为input，而后进行运算。

因为上述特性，将shuffle依赖就必须分为两个阶段(stage)去作：

第一个阶段(stage)须要把结果shuffle到本地，例如reduceByKey，首先要聚合某个key的全部记录，才能进行下一步的reduce计算，这个汇聚的过程就是shuffle

第二个阶段(stage)则读入数据进行处理。

同一个stage里面的task是能够并发执行的，下一个stage要等前一个stage ready

（和mapreduce的reduce须要等map过程ready 一脉相承）

(为何要写在本地：后面的RDD多个partition都要去读这个信息，若是放到内存，若是出现数据丢失，后面的全部步骤所有不能进行，违背了以前所说的须要parent RDD partition数据所有ready的原则。为何要保证parent RDD要ready，以下例，若是有一个partition未生成或者在内存中丢失，那么直接致使计算结果是彻底错误的：

写到文件中更加可靠。Shuffle会生成大量临时文件，以避免错误时从新计算，其使用的本地磁盘目录由spark.local.dir指定，缓存到磁盘的RDD数据。最好将这个属性设定为访问速度快的本地磁盘。能够配置多个路径到多个磁盘，增长IO带宽

在Spark 1.0 之后，SPARK_LOCAL_DIRS(Standalone, Mesos) or LOCAL_DIRS (YARN)参数会覆盖这个配置。好比Spark On YARN的时候，Spark Executor的本地路径依赖于Yarn的配置，而不取决于这个参数。)

对于transformation操做，以shuffle依赖为分隔，分为不一样的Stages。

窄依赖------>tasks会归并在同一个stage中，（相同节点上的task运算能够像pipeline同样顺序执行，不一样节点并行计算，互不影响）

shuffle依赖------>先后拆分为两个stage，前一个stage写完文件后下一个stage才能开始

action操做------>和其余tasks会归并在同一个stage（在没有shuffle依赖的状况下，生成默认的stage，保证至少一个stage）

小实验验证

例一：

def main(args: Array[String]): Unit = {
    val sp = new SparkConf();
    sp.setAppName("zhangzeli")
    sp.setMaster("local")
    val sc = new SparkContext(sp);
    val rdd =sc.parallelize(Array(1,2,3,4));//由于分的资源是两个核，因此默认设置为两个partition
    val cont = rdd.count();
    while (true){}

  }

Count是一个action操做。一个action会触发一个job，Count()这个action在整个job没有stage的状况下会生成一个默认的stage

结果：一个job，一个stage，两个task（由于有两个partition）

例二：

最终这个生成一个job，由于reducebykey是shuffle依赖，因此这里划分为两个stage

parallelize和map被分在一块儿，为stage0，map最后进行了ShuffleWrite

reduceByKey和count（）被划分到一个stage1里面了，最开始要进行shuffle read
Stage0的tasks以下图，两个partitions（两个tasks）都进行了shuffle write。两个task互相独立，并不须要依赖彼此作完或者怎样，因此他们在一个stage里面并发执行

Stage1的tasks以下：Stage1是依赖以前的stage0完成shuffle的，reduceByKey开始须要ShuffleRead stage0的计算结果

若是后面还有其余操做，这些操做是要等上面这个shuffle执行完的 reduceByKey 则在下一阶段，shuffleRead读到数据 因此根据shuffle依赖必须分为多个stage 但一个stage内部，多个task（partition）是独立并发执行的，互不打扰