spark系列10：spark底层运行原理，执行计划

绝地灬酷狼 2022-10-22 10:45 29阅读 0赞

## 1.Spark 底层逻辑 ##

导读

1.  从部署图了解 `Spark` 部署了什么, 有什么组件运行在集群中
2.  通过对 `WordCount` 案例的解剖, 来理解执行逻辑计划的生成
3.  通过对逻辑执行计划的细化, 理解如何生成物理计划

<table> 
 <tbody> 
  <tr> 
   <td>&nbsp;</td> 
   <td> <p>如无特殊说明, 以下部分均针对于&nbsp;<code>Spark Standalone</code>&nbsp;进行介绍</p> </td> 
  </tr> 
 </tbody> 
</table>

部署情况

在 `Spark` 部分的底层执行逻辑开始之前, 还是要先认识一下 `Spark` 的部署情况, 根据部署情况, 从而理解如何调度.

![WX20190513 233552][]

针对于上图, 首先可以看到整体上在集群中运行的角色有如下几个:

*  `Master Daemon`
    
    负责管理 `Master` 节点, 协调资源的获取, 以及连接 `Worker` 节点来运行 `Executor`, 是 Spark 集群中的协调节点
 *  `Worker Daemon`
    
    `Workers` 也称之为叫 `Slaves`, 是 Spark 集群中的计算节点, 用于和 Master 交互并管理 `Executor`.
    
    当一个 `Spark Job` 提交后, 会创建 `SparkContext`, 后 `Worker` 会启动对应的 `Executor`.
 *  `Executor Backend`
    
    上面有提到 `Worker` 用于控制 `Executor` 的启停, 其实 `Worker` 是通过 `Executor Backend` 来进行控制的, `Executor Backend` 是一个进程(是一个 `JVM` 实例), 持有一个 `Executor` 对象

另外在启动程序的时候, 有三种程序需要运行在集群上:

*  `Driver`
    
    `Driver` 是一个 `JVM` 实例, 是一个进程, 是 `Spark Application` 运行时候的领导者, 其中运行了 `SparkContext`.
    
    `Driver` 控制 `Job` 和 `Task`, 并且提供 `WebUI`.
 *  `Executor`
    
    `Executor` 对象中通过线程池来运行 `Task`, 一个 `Executor` 中只会运行一个 `Spark Application` 的 `Task`, 不同的 `Spark Application` 的 `Task` 会由不同的 `Executor` 来运行

案例

因为要理解执行计划, 重点不在案例, 所以本节以一个非常简单的案例作为入门, 就是我们第一个案例 WordCount

val sc = ...
    
    val textRDD = sc.parallelize(Seq("Hadoop Spark", "Hadoop Flume", "Spark Sqoop"))
    val splitRDD = textRDD.flatMap(_.split(" "))
    val tupleRDD = splitRDD.map((_, 1))
    val reduceRDD = tupleRDD.reduceByKey(_ + _)
    val strRDD = reduceRDD.map(item => s"${item._1}, ${item._2}")
    
    println(strRDD.toDebugString)
    strRDD.collect.foreach(item => println(item))

整个案例的运行过程大致如下:

1.  通过代码的运行, 生成对应的 `RDD` 逻辑执行图
2.  通过 `Action` 操作, 根据逻辑执行图生成对应的物理执行图, 也就是 `Stage` 和 `Task`
3.  将物理执行图运行在集群中

逻辑执行图

对于上面代码中的 `reduceRDD` 如果使用 `toDebugString` 打印调试信息的话, 会显式如下内容

(6) MapPartitionsRDD[4] at map at WordCount.scala:20 []
     |  ShuffledRDD[3] at reduceByKey at WordCount.scala:19 []
     +-(6) MapPartitionsRDD[2] at map at WordCount.scala:18 []
        |  MapPartitionsRDD[1] at flatMap at WordCount.scala:17 []
        |  ParallelCollectionRDD[0] at parallelize at WordCount.scala:16 []

根据这段内容, 大致能得到这样的一张逻辑执行图

![20190515002803][]

其实 RDD 并没有什么严格的逻辑执行图和物理执行图的概念, 这里也只是借用这个概念, 从而让整个 RDD 的原理可以解释, 好理解.

对于 RDD 的逻辑执行图, 起始于第一个入口 RDD 的创建, 结束于 Action 算子执行之前, 主要的过程就是生成一组互相有依赖关系的 RDD, 其并不会真的执行, 只是表示 RDD 之间的关系, 数据的流转过程.

物理执行图

当触发 Action 执行的时候, 这一组互相依赖的 RDD 要被处理, 所以要转化为可运行的物理执行图, 调度到集群中执行.

因为大部分 RDD 是不真正存放数据的, 只是数据从中流转, 所以, 不能直接在集群中运行 RDD, 要有一种 Pipeline 的思想, 需要将这组 RDD 转为 Stage 和 Task, 从而运行 Task, 优化整体执行速度.

以上的逻辑执行图会生成如下的物理执行图, 这一切发生在 Action 操作被执行时.

![20190515235205][]

从上图可以总结如下几个点

*  ![20190515235442][] 在第一个 `Stage` 中, 每一个这样的执行流程是一个 `Task`, 也就是在同一个 Stage 中的所有 RDD 的对应分区, 在同一个 Task 中执行
 *  Stage 的划分是由 Shuffle 操作来确定的, 有 Shuffle 的地方, Stage 断开

### 1.1. 逻辑执行图生成 ###

导读

1.  如何生成 RDD
2.  如何控制 RDD 之间的关系

6.1.1. RDD 的生成

重点内容

本章要回答如下三个问题

*  如何生成 RDD
 *  生成什么 RDD
 *  如何计算 RDD 中的数据

明确逻辑计划的边界

在 `Action` 调用之前, 会生成一系列的 `RDD`, 这些 `RDD` 之间的关系, 其实就是整个逻辑计划

例如上述代码, 如果生成逻辑计划的, 会生成如下一些 `RDD`, 这些 `RDD` 是相互关联的, 这些 `RDD` 之间, 其实本质上生成的就是一个 **计算链**

![20190519000019][]

接下来, 采用迭代渐进式的方式, 一步一步的查看一下整体上的生成过程

`textFile` 算子的背后

研究 `RDD` 的功能或者表现的时候, 其实本质上研究的就是 `RDD` 中的五大属性, 因为 `RDD` 透过五大属性来提供功能和表现, 所以如果要研究 `textFile` 这个算子, 应该从五大属性着手, 那么第一步就要看看生成的 `RDD` 是什么类型的 `RDD`

1.  `textFile` 生成的是 `HadoopRDD`
    
    ![20190519202310][]
    
    ![20190519202411][]
    
    <table> 
     <tbody> 
      <tr> 
       <td>&nbsp;</td> 
       <td> <p>除了上面这一个步骤以外, 后续步骤将不再直接基于代码进行讲解, 因为从代码的角度着手容易迷失逻辑, 这个章节的初心有两个, 一个是希望大家了解 Spark 的内部逻辑和原理, 另外一个是希望大家能够通过本章学习具有代码分析的能力</p> </td> 
      </tr> 
     </tbody> 
    </table>
2.  `HadoopRDD` 的 `Partitions` 对应了 `HDFS` 的 `Blocks`
    
    ![20190519203211][]
    
    其实本质上每个 `HadoopRDD` 的 `Partition` 都是对应了一个 `Hadoop` 的 `Block`, 通过 `InputFormat` 来确定 `Hadoop` 中的 `Block` 的位置和边界, 从而可以供一些算子使用
3.  `HadoopRDD` 的 `compute` 函数就是在读取 `HDFS` 中的 `Block`
    
    本质上, `compute` 还是依然使用 `InputFormat` 来读取 `HDFS` 中对应分区的 `Block`
4.  `textFile` 这个算子生成的其实是一个 `MapPartitionsRDD`
    
    `textFile` 这个算子的作用是读取 `HDFS` 上的文件, 但是 `HadoopRDD` 中存放是一个元组, 其 `Key` 是行号, 其 `Value` 是 `Hadoop` 中定义的 `Text` 对象, 这一点和 `MapReduce` 程序中的行为是一致的
    
    但是并不适合 `Spark` 的场景, 所以最终会通过一个 `map` 算子, 将 `(LineNum, Text)` 转为 `String` 形式的一行一行的数据, 所以最终 `textFile` 这个算子生成的 `RDD` 并不是 `HadoopRDD`, 而是一个 `MapPartitionsRDD`

`map` 算子的背后

![20190519101943][]

*  `map` 算子生成了 `MapPartitionsRDD`
    
    由源码可知, 当 `val rdd2 = rdd1.map()` 的时候, 其实生成的新 `RDD` 是 `rdd2`, `rdd2` 的类型是 `MapPartitionsRDD`, 每个 `RDD` 中的五大属性都会有一些不同, 由 `map` 算子生成的 `RDD` 中的计算函数, 本质上就是遍历对应分区的数据, 将每一个数据转成另外的形式
 *  `MapPartitionsRDD` 的计算函数是 `collection.map( function )`
    
    真正运行的集群中的处理单元是 `Task`, 每个 `Task` 对应一个 `RDD` 的分区, 所以 `collection` 对应一个 `RDD` 分区的所有数据, 而这个计算的含义就是将一个 `RDD` 的分区上所有数据当作一个集合, 通过这个 `Scala` 集合的 `map` 算子, 来执行一个转换操作, 其转换操作的函数就是传入 `map` 算子的 `function`
 *  传入 `map` 算子的函数会被清理
    
    ![20190519190306][]
    
    这个清理主要是处理闭包中的依赖, 使得这个闭包可以被序列化发往不同的集群节点运行

`flatMap` 算子的背后

![20190519190541][]

`flatMap` 和 `map` 算子其实本质上是一样的, 其步骤和生成的 `RDD` 都是一样, 只是对于传入函数的处理不同, `map` 是 `collect.map( function )` 而 `flatMap` 是 `collect.flatMap( function )`

从侧面印证了, 其实 `Spark` 中的 `flatMap` 和 `Scala` 基础中的 `flatMap` 其实是一样的

`textRDD` → `splitRDD` → `tupleRDD`

由 `textRDD` 到 `splitRDD` 再到 `tupleRDD` 的过程, 其实就是调用 `map` 和 `flatMap` 算子生成新的 `RDD` 的过程, 所以如下图所示, 就是这个阶段所生成的逻辑计划

![20190519211533][]

### **总结** ###

如何生成 `RDD` ?

生成 `RDD` 的常见方式有三种

*  从本地集合创建
 *  从外部数据集创建
 *  从其它 `RDD` 衍生

通过外部数据集创建 `RDD`, 是通过 `Hadoop` 或者其它外部数据源的 `SDK` 来进行数据读取, 同时如果外部数据源是有分片的话, `RDD` 会将分区与其分片进行对照

通过其它 `RDD` 衍生的话, 其实本质上就是通过不同的算子生成不同的 `RDD` 的子类对象, 从而控制 `compute` 函数的行为来实现算子功能

生成哪些 `RDD` ?

不同的算子生成不同的 `RDD`, 生成 `RDD` 的类型取决于算子, 例如 `map` 和 `flatMap` 都会生成 `RDD` 的子类 `MapPartitions` 的对象

如何计算 `RDD` 中的数据 ?

虽然前面我们提到过 `RDD` 是偏向计算的, 但是其实 `RDD` 还只是表示数据, 纵观 `RDD` 的五大属性中有三个是必须的, 分别如下

*  `Partitions List` 分区列表
 *  `Compute function` 计算函数
 *  `Dependencies` 依赖

虽然计算函数是和计算有关的, 但是只有调用了这个函数才会进行计算, `RDD` 显然不会自己调用自己的 `Compute` 函数, 一定是由外部调用的, 所以 `RDD` 更多的意义是用于表示数据集以及其来源, 和针对于数据的计算

所以如何计算 `RDD` 中的数据呢? 一定是通过其它的组件来计算的, 而计算的规则, 由 `RDD` 中的 `Compute` 函数来指定, 不同类型的 `RDD` 子类有不同的 `Compute` 函数

[WX20190513 233552]: /images/20221022/956f8a8a54c64e47bd37b4c274e31ec4.png
[20190515002803]: /images/20221022/29a2bdbb82dc41cc9aab6483e4af036d.png
[20190515235205]: /images/20221022/e49e0882399c4252b8cafa05740b546b.png
[20190515235442]: /images/20221022/f4cbf23e594348e99bb90fb4b1bf8c44.png
[20190519000019]: /images/20221022/1854482b915044fba81cd2075716762c.png
[20190519202310]: /images/20221022/ad5d33fc6bd4455da673390089ab64c0.png
[20190519202411]: /images/20221022/6f3dabc479ab4f4a917c77d967b6cf03.png
[20190519203211]: /images/20221022/adfe3526c59b41ea891ec65773fec5e9.png
[20190519101943]: /images/20221022/dc1d69f76e464fdb82b349eaccbd7b1a.png
[20190519190306]: /images/20221022/127a045b41e14f9d927b85af9448ddde.png
[20190519190541]: /images/20221022/7e00add78e3144b19603a4502ef06110.png
[20190519211533]: /images/20221022/e1a7cc6fb5bc46c188828408b04f331f.png