很楠不爱3

这个屌丝很懒，什么也没留下！

热门标签

Spark Core之Stage、Tank&Job划分_spark job划分

作者：很楠不爱3 | 2024-06-01 04:11:38

踩

spark job划分

Stage的划分那么要想清楚RDD的依赖关系，可点击下面了解

《RDD的依赖关系》

窄依赖，父RDD的分区最多只会被子RDD的一个分区使用，
宽依赖，父RDD的一个分区会被子RDD的多个分区使用（宽依赖指子RDD的每个分区都要依赖于父RDD的所有分区，这是shuffle类操作）

一、RDD的Job划分

任何一个Action算子就是一个job，因为每一个Action算子都会调用runjob方法

任务包括:ResultTask(最后一个阶段的任务) + ShuffleMapTask(非最后一个阶段)

二、RRD的Job生成和提交的四给阶段

四个步骤:

1,构建DAG
        用户提交的job将首先被转换成一系列RDD并通过RDD之间的依赖关系构建DAG,然后将DAG提交到调度系统;
        DAG描述多个RDD的转换过程，任务执行时，可以按照DAG的描述,执行真正的计算;
        DAG是有边界的:开始(通过sparkcontext创建的RDD)，结束 (触发action, 调用runjob就是一个完整的DAG形成了，一旦触发action,就形成了- -个完整的DAG) ;
        一个RDD描述了数据计算过程中的一个环节，而一个DAG包含多个RDD,描述了数据计算过程中的所有环节;
        一个spark application可以包含多个DAG,取决于具体有多少个action。
2，DAGScheduler将DAG切分stage (切分依据是shuffle) ,将stage中生成的task以taskset的形式发送给 TaskScheduler为什么要切分stage?
        一个复杂是业务逻辑(将多台机器上具有相同属性的数据聚合到一台机器上:shuffle)如果有shuffle,那么就意味着前面阶段产生结果后，才能执行下一-个阶段,下一个阶段的计算依赖上一个阶段的数据在同一个stage中，会有多个算子，可以合并到一-起，我们很难” 称其为pipeline (流水线,严格按照流程、顺序执行)

3，TaskScheduler 调度task (根据资源情况将task调度到Executors)

4，Executors接收task, 然后将task交给线程池执行。

三、Stage的划分

划分stage的过程:从最后一个RDD开始 ,调用递归算法查找该RDD的父RDD ,找到父RDD后开始遍历,判断父RDD和该RDD的依赖关系,如果是宽依赖,就把父RDD和前面所有RDD都划分一个stage ,如果是窄依赖,继续递归查找父RDD的父RDD ,递归的出口是直到找不到父RDD.最后把所有的RDD统一划分一个stage.

一个job有一个或多个Stage组成，一个Stage由一个或多个Task组成

下图从源文件到输出总共有4给RDD，textDile-RDD --> flaMap-RDD --> map-RDD都是窄依赖所以划分为一个Stage, 然后reduceByKey-RDD这是宽依赖，会发生shuffle，这里段开，重新一个Stage，后面依照这个来划分Stage

划分规则

从后向前推理，遇到宽依赖就断开，遇到窄依赖就把当前的RDD加⼊到Stage中；
每个Stage⾥⾯的Task的数量是由该Stage中最后⼀个RDD的Partition数量决定的；
最后⼀个Stage⾥⾯的任务的类型是ResultTask，前⾯所有其他Stage⾥⾯的任务类型都是ShuffleMapTask；
代表当前Stage的算⼦⼀定是该Stage的最后⼀个计算步骤；

总结：由于 spark 中 stage 的划分是根据 shuffle 来划分的，⽽宽依赖必然有 shuffle 过程，因此可以说 spark 是根据宽窄依赖来划分stage 的。

为什么要stage划分?

在shulle之前保存数据用于重复计算
还可以提高并行度

四、Task划分

输⼊可能以多个⽂件的形式存储在HDFS 上，每个 File 都包含了很多块，称为 Block 。

当Spark 读取这些⽂件作为输⼊时，会根据具体数据格式对应的 InputFormat 进⾏解析，⼀般是将若⼲个 Block 合并成⼀个输⼊分⽚，称为InputSplit ，注意 InputSplit 不能跨越⽂件。

随后将为这些输⼊分⽚⽣成具体的 Task 。 InputSplit 与 Task 是⼀⼀对应的关系。

随后这些具体的 Task 每个都会被分配到集群上的某个节点的某个 Executor 去执⾏。

每个节点可以起⼀个或多个Executor。
每个Executor由若⼲core组成，每个Executor的每个core⼀次只能执⾏⼀个Task。
每个Task执⾏的结果就是⽣成了⽬标RDD的⼀个partiton。

注意 : 这⾥的 core 是虚拟的 core ⽽不是机器的物理 CPU 核，可以理解为就是 Executor 的⼀个⼯作线程。

⽽ Task 被执⾏的并发度 = Executor 数⽬（ SPARK_EXECUTOR_INSTANCES ） * 每个 Executor 核数（SPARK_EXECUTOR_CORES ）

Task的数量由Partition影响,与算子数据无关.只要定义在一个Stage的算子,都是一个分区(任务)上的操作.

总结： RDD 在计算的时候，每个分区都会起⼀个 task ，所以 rdd 的分区数⽬决定了总的的 task 数⽬。

五、WebUI中查看Stage和Task

在 spark-shell 中执⾏ wordcount

sc.textFile("hdfs://hadoop01:9000/wc").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect

因为在计算的时候 textFile 默认是 2 个 partition, 整个计算流程是 3 个 stage, 实际得到的 task 应该会是 6 个 , 实际的个数是 4个

要是出现 skipped 那么就会减少对应的 task, 但是这是没有问题的并且是对

任务出现 skipped 是正常的，之所以出现 skipped 是因为要计算的数据已经缓存到了内存，没有必要再重复计算。

出现 skipped 对结果没有影响 , 并且也是⼀种计算优化

在发⽣ shuffle 的过程中，会发⽣ shuffle write 和 shuffle read 。

shuffle write ：发⽣在 shuffle 之前，把要 shuffle 的数据写到磁盘

为什么：为了保证数据的安全性、避免占⽤⼤量的内存

shuffle read ：发⽣在 shuffle 之后，下游 RDD 读取上游 RDD 的数据的过程

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/很楠不爱3/article/detail/655885