当前位置:   article > 正文

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

spark job划分

目录

  《RDD的依赖关系》

一、RDD的Job划分

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

三、Stage的划分

 四、Task划分

 五、WebUI中查看Stage和Task


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

划分规则
  1. 从后向前推理,遇到宽依赖就断开,遇到窄依赖就把当前的RDD加⼊到Stage中;
  2. 每个Stage⾥⾯的Task的数量是由该Stage中最后 ⼀个RDDPartition数量决定的;
  3. 最后⼀个Stage⾥⾯的任务的类型是ResultTask,前⾯所有其他Stage⾥⾯的任务类型都是ShuffleMapTask
  4. 代表当前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中查看StageTask

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
推荐阅读
相关标签
  

闽ICP备14008679号