Spark基础 DAG

叁歲伎倆 2023-06-30 08:39 13阅读 0赞

为什么使用spark的原因是早期的编程模式MapReduce缺乏对数据共享的高效元语，会造成磁盘I/O 以及序列号等开销，spark提出了统一的编程抽象---弹性分布式数据集（RDD）,该模型可以令并行计算阶段间高效地进行数据共享。spark处理数据时，会将计算转化为一个有向无环图(DAG)的任务集，RDD能够有效的恢复DAG中故障和慢节点执行的任务，并且RDD提供一种基于粗粒度变换的接口，记录创建数据集的“血统”，能够实现高效的容错性。

spark的作业和任务调度系统是其核心，它能够有效的进行调度的根本原因是因为对任务划分DAG和容错。

DAG，有向无环图，Directed Acyclic Graph的缩写，常用于建模。Spark中使用DAG对RDD的关系进行建模，描述了RDD的依赖关系，这种关系也被称之为lineage，RDD的依赖关系使用Dependency维护，参考Spark RDD之Dependency，DAG在Spark中的对应的实现为DAGScheduler。

基础概念  
介绍DAGScheduler中的一些概念，有助于理解后续流程。  
作业（Job）  调用RDD的一个action，如count，即触发一个Job，spark中对应实现为ActiveJob，DAGScheduler中使用集合activeJobs和jobIdToActiveJob维护Job  
调度阶段（Stage ）   代表一个Job的DAG，会在发生shuffle处被切分，切分后每一个部分即为一个Stage，Stage实现分为ShuffleMapStage和ResultStage，一个Job切分的结果是0个或多个ShuffleMapStage加一个ResultStage  
任务（Task ）   最终被发送到Executor执行的任务，和stage的ShuffleMapStage和ResultStage对应，其实现分为ShuffleMapTask和ResultTask

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5ndmFsdWU_size_16_color_FFFFFF_t_70][]  
1) 一个RDD生成两个RDD：

RDD2 = RDD1.filter(xxxxx)

RDD3 = RDD1.filter(yyyy)

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

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

则是RDD2 RDD3变成RDD4的过程

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

是从RDD4到RDD5这样的过程

上述都是transformation

RDD5.collect();  //action

RDD5.foreach();  //action

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

（可以和上一章中Persist中的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）  计算就涉及到节点间网络传输

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5ndmFsdWU_size_16_color_FFFFFF_t_70 1][]

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

子分区，来自部分父分区                         子分区，来自全部父分区

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。

如图所示，左边的都是右边的父分区![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5ndmFsdWU_size_16_color_FFFFFF_t_70 2][]

（三）、划分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未生成或者在内存中丢失，那么直接导致计算结果是完全错误的：![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5ndmFsdWU_size_16_color_FFFFFF_t_70 3][]

写到文件中更加可靠。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）。

（四）、小实验验证  
例一：

在spark-shell里面跑小程序，然后在yarn里面观察

val rdd =  sc.parallelize(Array(1,2,3,7)) （因为分的资源是两个核，所以默认设置为两个partition）

rdd.count()

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

结果：一个job，一个stage，两个task（因为有两个partition）

> 例二：
> 
> scala> val rdd= sc.parallelize(Array("hello","spark","hello","zhangvalue"))  
> rdd: org.apache.spark.rdd.RDD\[String\] = ParallelCollectionRDD\[11\] at parallelize at <console>:24  
>    
> scala> val pairs = rdd.map(s=>(s,1))  
> pairs: org.apache.spark.rdd.RDD\[(String, Int)\] = MapPartitionsRDD\[12\] at map at <console>:25  
>    
> scala> val count = pairs.reduceByKey((a,b) => a+b)  
> count: org.apache.spark.rdd.RDD\[(String, Int)\] = ShuffledRDD\[13\] at reduceByKey at <console>:25  
>    
> scala> count.count()  
> res4: Long = 3

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

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

reduceByKey和count（）被划分到一个stage1里面了，最开始要进行shuffle read

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5ndmFsdWU_size_16_color_FFFFFF_t_70 4][]  
Stage0的tasks如下图，两个partitions（两个tasks）都进行了shuffle write。两个task互相独立，并不需要依赖彼此做完或者怎样，所以他们在一个stage里面并发执行

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

如果后面还有其他操作，这些操作是要等上面这个shuffle执行完的

reduceByKey 则在下一阶段，shuffleRead读到数据![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5ndmFsdWU_size_16_color_FFFFFF_t_70 5][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5ndmFsdWU_size_16_color_FFFFFF_t_70 6][]![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5ndmFsdWU_size_16_color_FFFFFF_t_70 5][]

所以根据shuffle依赖必须分为多个stage

但一个stage内部，多个task（partition）是独立并发执行的，互不打扰

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5ndmFsdWU_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20181203221302906.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5ndmFsdWU=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5ndmFsdWU_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20181203221414320.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5ndmFsdWU=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5ndmFsdWU_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20181203221447601.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5ndmFsdWU=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5ndmFsdWU_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20181203221513992.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5ndmFsdWU=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5ndmFsdWU_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20181203222109577.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5ndmFsdWU=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5ndmFsdWU_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20181203223859978.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5ndmFsdWU=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5ndmFsdWU_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20181203224029385.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5ndmFsdWU=,size_16,color_FFFFFF,t_70