spark中job，stage，task之间的关系

我就是我 2022-12-20 06:14 241阅读 0赞

# 1. **什么是job** #

Job简单讲就是提交给spark的任务。

# 2. **什么是stage** #

Stage是每一个job处理过程要分为的几个阶段。

# **3什么是task** #

Task是每一个job处理过程要分几为几次任务。Task是任务运行的最小单位。最终是要以task为单位运行在executor中。

# 3. **Job和stage和task之间有什么关系** #

Job----> 一个或多个stage---> 一个或多个task

job:应用程序中的每个acion操作（如collect、count、saves、reduce）都会创建成一个job，一个job由多个stage和task组成。

每个job又会根据是否需要数据混排生成一个或多个stage。  
stage的数量取决于程序中需要进行数据混排（shuffle）的操作的数量,stage会根据RDD分区的数量生成相对应数量的task。

stage切割规则：**从后往前**，**遇到宽依赖（shuffle或洗牌，算子如：**groupByKey、countByKey、reduceByKey、join、groupBy**）就切割****stage。**

![0f60e2073d063bcca78c289d055fed60.png][]  
task的内容和stage完全相同，当分区数量为n时，会有n个相同效果的task被分发到执行程序中执行。数量取决于job、stage和文件分块数的乘积。  
taskNum=jobNum\**stageNum\**blockNum

提高stage的并行度

**1）Stage的task并行度是由stage的最后一个RDD的分区数来决定的 。一般来说，一个partiotion对应一个task,但最后reduce的时候可以手动改变reduce的个数，也就是分区数，即改变了并行度。例如reduceByKey(XXX,3),GroupByKey(4)，union由的分区数由前面的相加。**

2） **如何提高stage的并行度**：reduceBykey(xxx,numpartiotion),join(xxx,numpartiotion)

**下图是一个job分成了三个stage：**

![70][]

# **5.一个stage的task的数量是有谁来决定的？** #

是由输入文件的切片个数来决定的。在HDFS中不大于128m的文件算一个切片（默认128m）。通过算子修改了某一个rdd的分区数量，task数量也会同步修改。

# **6.一个job任务的task数量是由谁来决定的？** #

一个job任务可以有一个或多个stage，一个stage又可以有一个或多个task。所以一个job的task数量是  （stage数量 \* task数量）的总和。

![70 1][]

# 上图就是job1有3个task。 #

# **7.每一个stage中的task最大的并行度？** #

并行度：是指指令[并行执行][Link 1]的最大条数。在指令流水中，同时执行多条指令称为指令并行。

理论上：每一个stage下有多少的分区，就有多少的task，task的数量就是我们任务的最大的并行度。

（一般情况下，我们一个task运行的时候，使用一个cores）

实际上：最大的并行度，取决于我们的application任务运行时使用的executor拥有的cores的数量。

![70 2][]

如图所示，cores为3，那么最大的并行度就是3。

# **8.如果我们的task数量超过这个cores的总数怎么办？** #

先执行cores个数量的task，然后等待cpu资源空闲后，继续执行剩下的task。

![70 3][]

# **9.spark执行时读条中的内容讲解** #

![70 4][]

satge11：  当前的satge编号

（5+4）/9:

9:当前stage的task的数量，5：已完成的task数量，4：等待执行的task数量。

# 10 宽窄依赖 #

RDD之间有一系列的依赖关系，依赖关系又分为窄依赖和宽依赖。Spark中的Stage其实就是一组并行的任务，任务是一个个的task 。

*  **窄依赖**

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

*  **宽依赖**

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

窄依赖：可以理解为独生子女

宽依赖：可以理解为超生

如上图所示，一个文件被textFile后经过两次转换，每次转换后生成的新的RDD都和原分区一一对应，为窄依赖关系。

当map算子计算失败时，可以直接去上一步的RDD中重新计算，而不需要其他的分区参与。但是如果最后的action算子在计算某个RDD时，计算失败，将会到上一层的多个分区中寻找依赖关系，比较麻烦。

**其实区分宽窄依赖主要就是看父RDD的一个Partition的流向，要是流向一个的话就是窄依赖，流向多个的话就是宽依赖。看图理解：**

**![fc84ddc159512b8c259011585a0b4dd9.png][]**

*  **Stage概念**

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

*  **stage切割规则**

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

**![0f60e2073d063bcca78c289d055fed60.png][]**

*  **stage计算模式**

pipeline管道计算模式,pipeline只是一种计算思想，模式。

![7fc1f88763bd51c8c5c65f3098ac46e4.png][]

备注：图中几个理解点：

1、Spark的pipeLine的计算模式，相当于执行了一个高阶函数f3(f2(f1(textFile))) !+!+!=3 也就是来一条数据然后计算一条数据，把所有的逻辑走完，然后落地，准确的说一个task处理遗传分区的数据 因为跨过了不同的逻辑的分区。而MapReduce是 1+1=2,2+1=3的模式，也就是计算完落地，然后在计算，然后再落地到磁盘或内存，最后数据是落在计算节点上，按reduce的hash分区落地。所以这也是比Mapreduce快的原因，完全基于内存计算。

2、管道中的数据何时落地：**shuffle write的时候，对RDD进行持久化的时候。**

3.**   Stage的task并行度是由stage的最后一个RDD的分区数来决定的 。一般来说，一个partiotion对应一个task,但最后reduce的时候可以手动改变reduce的个数，也就是分区数，即改变了并行度。例如reduceByKey(XXX,3),GroupByKey(4)，union由的分区数由前面的相加。**

4.、**如何提高stage的并行度**：reduceBykey(xxx,numpartiotion),join(xxx,numpartiotion)

# #

**shuffle 是划分 DAG 中 stage 的标识,同时影响 Spark 执行速度的关键步骤**.   
　　RDD 的 Transformation 函数中,又分为窄依赖(narrow dependency)和宽依赖(wide dependency)的操作.窄依赖跟宽依赖的区别是**是否发生 shuffle(洗牌) 操作**.宽依赖会发生 shuffle 操作. 窄依赖是子 RDD的各个分片(partition)不依赖于其他分片,能够独立计算得到结果,宽依赖指子 RDD 的各个分片会依赖于父RDD 的多个分片,所以会造成父 RDD 的各个分片在集群中重新分片, 看如下两个示例:

1.  `// Map: "cat" -> c, cat`
2.  `val rdd1 = rdd.Map(x => (x.charAt(0), x))`
3.  `// groupby same key and count`
4.  `val rdd2 = rdd1.groupBy(x => x._1).`
5.  `Map(x => (x._1, x._2.toList.length))`

第一个 Map 操作将 RDD 里的各个元素进行映射, RDD 的各个数据元素之间不存在依赖,可以在集群的各个内存中独立计算,也就是**并行化**,第二个 groupby 之后的 Map 操作,为了计算相同 key 下的元素个数,需要把相同 key 的元素聚集到同一个 partition 下,所以造成了数据在**内存中的重新分布**,即 shuffle 操作.**shuffle 操作是 spark 中最耗时的操作,应尽量避免不必要的 shuffle**.   
　　宽依赖主要有两个过程: shuffle write 和 shuffle fetch. 类似 Hadoop 的 Map 和 Reduce 阶段.shuffle write 将 ShuffleMapTask 任务产生的中间结果缓存到内存中, shuffle fetch 获得 ShuffleMapTask 缓存的中间结果进行 ShuffleReduceTask 计算,**这个过程容易造成OutOfMemory**.   
　　shuffle 过程内存分配使用 ShuffleMemoryManager 类管理,会针对每个 Task 分配内存,Task 任务完成后通过 Executor 释放空间.这里可以把 Task 理解成不同 key 的数据对应一个 Task. **早期**的内存分配机制使用公平分配,即不同 Task 分配的内存是一样的,但是这样容易造成内存需求过多的 Task 的 OutOfMemory, 从而造成多余的 磁盘 IO 过程,影响整体的效率.(**例:**某一个 key 下的数据明显偏多,但因为大家内存都一样,这一个 key 的数据就容易 OutOfMemory).**1.5版以后** Task 共用一个内存池,内存池的大小默认为 JVM 最大运行时内存容量的16%,分配机制如下:假如有 N 个 Task,ShuffleMemoryManager 保证每个 Task 溢出之前至少可以申请到1/2N 内存,且至多申请到1/N,N 为当前活动的 shuffle Task 数,因为N 是一直变化的,所以 manager 会一直追踪 Task 数的变化,重新计算队列中的1/N 和1/2N.但是这样仍然容易造成内存需要多的 Task 任务溢出,所以最近有很多**相关的研究是针对 shuffle 过程内存优化的**.   
![@shuffle 流程 | center | 520X400][shuffle _ _ center _ 520X400]

[0f60e2073d063bcca78c289d055fed60.png]: /images/20221120/83ec1e4122a64629b4650fd49fee983a.png
[70]: /images/20221120/ac6d4b3a1142474ba50a2b1c4766eb74.png
[70 1]: /images/20221120/5a5186e0485d44939cf19b9cc4baee76.png
[Link 1]: https://baike.so.com/doc/6555303-6769052.html
[70 2]: /images/20221120/0b44cda3774c48eeb177861218373d9f.png
[70 3]: https://img-blog.csdn.net/20180630120440664?watermark/2/text/aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L215c18zNTA4OA==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70
[70 4]: https://img-blog.csdn.net/20180630120447616?watermark/2/text/aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L215c18zNTA4OA==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70
[fc84ddc159512b8c259011585a0b4dd9.png]: https://img-blog.csdnimg.cn/img_convert/fc84ddc159512b8c259011585a0b4dd9.png
[7fc1f88763bd51c8c5c65f3098ac46e4.png]: https://img-blog.csdnimg.cn/img_convert/7fc1f88763bd51c8c5c65f3098ac46e4.png
[shuffle _ _ center _ 520X400]: https://img-blog.csdn.net/20161103173627340