Spark-stage、task发送、资源任务调度、算子

墨蓝 2021-12-01 14:58 432阅读 0赞

# Spark-stage、task发送、资源任务调度、算子 #

### spark中job/stage/RDD/partition/task之间的对应关系 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70]  
**Tips:**

\`\`一个applicition中有多个job job数量由Action算子决定 job间顺序执行

\`\`一个job对应多个stage stage数量由宽依赖个数决定 stage间顺序执行

\`\`一个stage有多个RDD

\`\`一个RDD有多个partition(计算空间内最小单元/RDD的最小单元)

RDD是由分布在各个节点上的partition组成的

同一份数据（RDD）的partition 大小不一，数量不定

\`\`一个stage有多个可并行的task(数据逻辑处理)

\`\`一个task对应一个partition

task在内存资源足够的情况下是并行的，当内存资源不足时是串行的

partition数默认是和block块数是一致，但是后期shuffle过程是可变的

partition由application里的算子和最初读入的数据分块数量决定

partition 是弹性分布式数据集RDD的最小单元

block位于存储空间、partition 位于计算空间，  
 block的大小是固定的、partition 大小是不固定的，  
 block是有冗余的、不会轻易丢失，partition（RDD）没有冗余设计、丢失之后重新计算得到

## 一、stage ##

### 1概念 ###

任务层面：Application -> job -> stage -> tasks=partition(RDD最小处理单元)
    资源层面：Master -> Worker -> Executor -> Threadpool

Spark任务会根据RDD之间的依赖关系，形成一个DAG有向无环图，DAG会提交给DAGScheduler，DAGScheduler会把DAG划分相互依赖的多个stage，划分stage的依据就是RDD之间的宽窄依赖。

遇到宽依赖就划分stage,每个stage包含一个或多个task任务。然后将这些task以taskSet的形式提交给TaskScheduler运行。

stage是由一组并行的task组成。

### 2切割规则 ###

切割规则：从后往前，遇到宽依赖就切割stage。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70 1]

### 3-1计算模式-Pipeline管道模式 ###

以wordCount为例：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70 2]

注意：

管道模式处理方式 递归 fn（fn-1（fn-1（…））），在默认情况下数据不会落地

stage的并行度 由落地前最后一个RDD的partition决定，但是其余RDD都可以影响最终RDD的个数

管道中数据落地 进行shuffle.write操作/对某一RDD进行持久化

组成。

### 3-2计算模式-Pipeline管道模式 ###

以wordCount为例：

\[外链图片转存中…(img-PZ4aspmI-1563543674924)\]

注意：

管道模式处理方式 递归 fn（fn-1（fn-1（…））），在默认情况下数据不会落地

stage的并行度 由落地前最后一个RDD的partition决定，但是其余RDD都可以影响最终RDD的个数

管道中数据落地 进行shuffle.write操作/对某一RDD进行持久化

提高stage并行度 手动设置partition数量，distinct(去重)，xxxByKey等算子

## 二、task发送 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70 3]

Application -> job -> stage -> tasks

### 1组件 ###

Applicition N个job

RDD 一个job有N个有有向无环图

按照RDD间的宽窄依赖关系切分job，得到N个stage

DAGScheduler 切分stage，得到N个task  
 以TaskSet提交TaskSchduler

TaskSchduler 遍历TaskSet，将task发送到Worker节点

Worker 执行execute

### 2失败重新发送(2次) ###

Tasks 发送 Worker 失败，重发3次，若依旧失败则找 DAGScheduler 的相关的 stage

DAGScheduler 得到发送失败消息后，重新发送 TaskSet，失败4次后，停止该 job，停止application

注意：

DAGScheduler 确认发送消息失败后，在有问题的 task 后的 job 都会停止

但是之前执行完毕的 job 依旧是成功的

### 3推测执行机制 ###

在同一组 job 中，长时间除了一个 stage 任务执行缓慢没有完成，【】就会重启一个

注意：

由于推测执行机制会重新开始执行数据，所以会造成数据重复问题，因此推测执行机制与 ETL不相容

## 三、资源、任务调度 ##

### 1资源任务调度图 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70 4]

1-6 资源调度 7-11 任务调度

### 2粗粒度资源申请 ###

Spark基于粗粒度资源申请

application启动之间先将所有的资源申请完毕，如果申请不到就一直等待直到申请到为止。必须一次申请完毕之后才会执行任务，这样，task执行不需要自己申请资源加快了application的执行效率，必须最后一个task执行完毕后才会释放所有资源

优点：执行速度快

缺点：容易造成资源不能充分利用

### 3细粒度资源申请 ###

MP基于细粒度资源申请

application执行之前，不会讲所有task申请完毕，task执行时，task需要自己申请资源，释放资源，这样task执行相当慢，但是集群资源可以充分利用

优点：集群资源可以充分利用

缺点：application执行相对慢

\*\*Tips:\*\*个人理解

粗粒度：资源调度开一次，新task只开启新的任务调度 快但要等待，浪费资源

细粒度：一个task开一次资源调度+任务调度 慢但资源可用度高

## 四、算子 ##

### Transformations算子 ###

#### mapPartitionWithIndex-取分区索引号和该分区中的数据 ####

val rdd1: RDD[String] = sc.parallelize(List[String](
          "love1", "love2", "love3", "love4",
          "love5", "love6", "love7", "love8",
          "love9", "love10", "love11", "love12"
        ), 3)
    
        /**
          * mapPartitionsWithIndex(index,iter)
          * 拿到取分区索引号和该分区中的数据
          */
        //
        val rdd2: RDD[String] = rdd1.mapPartitionsWithIndex((index, iter) => {
          val list = new ListBuffer[String]()
          while (iter.hasNext) {
            val one = iter.next()
            list.+=(s"【$index】【$one】")
          }
          list.iterator
        })
    //    rdd2.foreach(println)

#### repartition-增加分区数 ####

/**
          * repartition
          * 控制RDD的分区数目，增减
          * 宽依赖，会产生suffle
          * 常用于增加分区
          */
        val  rdd3 = rdd2.repartition(2)
        val rdd4: RDD[String] = rdd3.mapPartitionsWithIndex((index, iter) => {
          val list = new ListBuffer[String]()
          while (iter.hasNext) {
            val one = iter.next()
            list.+=(s"【$index】【$one】")
          }
          list.iterator
        })
    //    val result = rdd4.collect().foreach(println)

#### coalesce-减少分区（常用） ####

/**
          * coalesce
          * 常用于减少分区
          * 控制RDD的分区数目，增减 窄依赖
          * 添加属性true  = repartition 变成宽依赖差生shuffle
          * 当coalesce 由少的分区分到多的分区时，在不让产生shuffle情况下，是没有作用的
          */
          val rdd5 = rdd2.coalesce(2)
    //    val  rdd5 = rdd2.coalesce(2,true)
        val rdd6: RDD[String] = rdd5.mapPartitionsWithIndex((index, iter) => {
          val list = new ListBuffer[String]()
          while (iter.hasNext) {
            val one = iter.next()
            list.+=(s"【$index】【$one】")
          }
          list.iterator
        })
        val result = rdd6.collect().foreach(println)

#### groupByKey-相同K放一起 ####

/**
      * groupByKey
      * 相同k数据放一起
      * (k,CompactBuffer(v1, v2))
      */
    val rdd1 = sc.parallelize(List[(String,Int)](("a",1),("a",2),("b",333),("b",444)))
    rdd1.groupByKey().foreach(println)

#### zip-1:1压缩集合元素 ####

val rdd1: RDD[String] = sc.parallelize(List[String]("zhangsan", "lisi", "wangwu", "maliu"))
    val rdd2: RDD[String] = sc.parallelize(List[String]("100", "200", "300", "400"))
    /**
      * zip
      * 1:!压缩元素，2集合元素个数必须一样
      */
    val result: RDD[(String, String)] = rdd1.zip(rdd2)
    result.foreach(println)

#### zipWithIndex-压缩数据和所在索引 ####

/**
      * zipWithIndex
      * 压缩集合元素和该元素的索引
      */
     val rdd1: RDD[String] = sc.parallelize(List[String]("zhangsan", "lisi", "wangwu", "maliu"))
    val result1: RDD[(String, Long)] = rdd1.zipWithIndex()
    result1.foreach(println)

### Action算子 ###

#### countByKey-统计(k,n) ####

必须作用在KV格式的数据上

val rdd1 = sc.parallelize(List[(String,Int)](
          ("a",100),
          ("a",100),
          ("a",200),
          ("b",2),
          ("c",300),
          ("d",100),
          ("d",500)
        ))
        val stringToLong: collection.Map[String, Long] = rdd1.countByKey()

#### countByValue-统计((k,v)n) ####

不必作用在kv格式上

将原集合的每个kv看做每一个整体，再统计其整体出现的次数

val rdd2 = sc.parallelize(List[(String,Int)](
          ("a",100),
          ("a",100),
          ("a",200),
          ("b",2),
          ("c",300),
          ("d",100),
          ("d",500)
        ))
        val tupleToLong: collection.Map[(String, Int), Long] = rdd2.countByValue()
        tupleToLong.foreach(println)
        
        
    
        val rdd3 = sc.parallelize(List[Int](1,2,3,4,4,4,5,2))
        val intToLong: collection.Map[Int, Long] = rdd3.countByValue()
        intToLong.foreach(println)

#### reduce-集合值相加 ####

/**
      * reduce
      * 值相加
      */
    val rdd = sc.parallelize(List[Int](1,2,3,4,5))
    //    val result: Int = rdd.reduce(_+_)
    val result: Int = rdd.reduce((v1, v2)=>{v1+v2})
    println(result)

#### reduce-集合所有值操作 ####

/**
      * reduce
      * 值相加
      */
    val rdd = sc.parallelize(List[Int](1,2,3,4,5))
    //    val result: Int = rdd.reduce(_+_)
    val result: Int = rdd.reduce((v1, v2)=>{v1+v2})
    println(result）

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70]: /images/20211201/bf16d4f9976149a3ace1986bf77ef950.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70 1]: /images/20211201/33ddcd9fca6e4516a0174f9b55585725.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70 2]: /images/20211201/9d333abdbe4143d6a1b3925ce1798472.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70 3]: /images/20211201/388ee280b7014f97b5124baf79b977e1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70 4]: /images/20211201/a75812733c3f4700b73e356789008668.png