Spark原理——物理执行图

物理执行图

• 物理图的作用是什么?

  问题一: 物理图的意义是什么?

  物理图解决的其实就是 RDD 流程生成以后, 如何计算和运行的问题, 也就是如何把 RDD 放在集群中执行的问题

  

  问题二: 如果要确定如何运行的问题, 则需要先确定集群中有什么组件

  • 首先集群中物理元件就是一台一台的机器
  • 其次这些机器上跑的守护进程有两种: Master, Worker
    • 每个守护进程其实就代表了一台机器, 代表这台机器的角色, 代表这台机器和外界通信
    • 例如我们常说一台机器是 Master, 其含义是这台机器中运行了一个 Master 守护进程, 如果一台机器运行了Master 的同时又运行了 Worker, 则说这台机器是 Master 也可以, 说它是 Worker 也行
  • 真正能运行 RDD 的组件是: Executor, 也就是说其实 RDD 最终是运行在 Executor 中的, 也就是说, 无论是 Master 还是 Worker 其实都是用于管理 Executor 和调度程序的

  结论： RDD 一定在 Executor 中计算, 而 Master 和 Worker 负责调度和管理 Executor

  问题三: 物理图的生成需要考虑什么问题?

  • 要计算 RDD, 不仅要计算, 还要很快的计算 → 优化性能
  • 要考虑容错, 容错的常见手段是缓存 → RDD 要可以缓存

  结论：在生成物理图的时候, 不仅要考虑效率问题, 还要考虑一种更合适的方式, 让 RDD 运行的更好

• 谁来计算 RDD ?

  问题一: RDD 是什么, 用来做什么 ?

  回顾一下 RDD 的五个属性

  • A list of partitions
  • A function for computing each split
  • A list of dependencies on other RDDs
  • Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)
  • Optionally, a list of preferred locations to compute each split on (e.g. block locations for an HDFS file)

  简单的说就是: 分区列表, 计算函数, 依赖关系, 分区函数, 最佳位置

  • 分区列表, 分区函数, 最佳位置, 这三个属性其实说的就是数据集在哪, 如何分区，在哪更合适
  • 计算函数和依赖关系, 这两个属性其实说的是数据集从哪来

  结论：RDD 是一个数据集的表示, 不仅表示了数据集, 还表示了这个数据集从哪来, 如何计算

  但是问题是, 谁来计算 ? 如果为一台汽车设计了一个设计图, 那么设计图自己生产汽车吗 ?

  问题二: 谁来计算 ?

  前面我们明确了两件事, RDD 在哪被计算? 在 Executor 中. RDD 是什么? 是一个数据集以及其如何计算的图纸.

  直接使用 Executor 也是不合适的, 因为一个计算的执行总是需要一个容器, 例如 JVM 是一个进程, 只有进程中才能有线程, 所以这个计算 RDD 的线程应该运行在一个进程中, 这个进程就是 Exeutor, Executor 有如下两个职责

  • 和 Driver 保持交互从而认领属于自己的任务

    

  • 接受任务后, 运行任务

    

  所以, 应该由一个线程来执行 RDD 的计算任务, 而 Executor 作为执行这个任务的容器, 也就是一个进程, 用于创建和执行线程, 这个执行具体计算任务的线程叫做 Task

  问题三: Task 该如何设计 ?

  第一个想法 是 每个 RDD 都由一个 Task 来计算

  第二个想法 是 让数据流动

  第三个想法 是 分阶段执行

  • 第一个想法: 为每个 RDD 的分区设置一组 Task

    

    大概就是每个 RDD 都有三个 Task, 每个 Task 对应一个 RDD 的分区, 执行一个分区的数据的计算

    但是这么做有一个非常难以解决的问题, 就是数据存储的问题, 例如 Task 1, 4, 7, 10, 13, 16 在同一个流程上, 但是这些 Task 之间需要交换数据, 因为这些 Task 可能被调度到不同的机器上, 所以 Task1 执行完了数据以后需要暂存, 后交给 Task4 来获取

    这只是一个简单的逻辑图, 如果是一个复杂的逻辑图, 会有什么表现? 要存储多少数据? 无论是放在磁盘还是放在内存中, 是不是都是一种极大的负担?

  • 第二个想法: 让数据流动

    很自然的, 第一个想法的问题是数据需要存储和交换, 那不存储不就好了吗? 对, 可以让数据流动起来

    第一个要解决的问题就是, 要为数据创建管道(Pipeline), 有了管道, 就可以流动

    

    简单来说, 就是为所有的 RDD 有关联的分区使用同一个 Task, 但是就没问题了吗? 请关注红框部分

    

    这两个 RDD 之间是 Shuffle 关系, 也就是说, 右边的 RDD 的一个分区可能依赖左边 RDD 的所有分区, 这样的话, 数据在这个地方流不动了, 怎么办?

  • 第三个想法: 划分阶段

    既然在 Shuffle 处数据流不动了, 那就可以在这个地方中断一下,后续讲解

  • 如何划分阶段 ?

    为了减少执行任务, 减少数据暂存和交换的机会, 所以需要创建管道, 让数据沿着管道流动, 其实也就是原先每个 RDD 都有一组 Task, 现在改为所有的 RDD 共用一组 Task, 但是也有问题, 问题如下

    

    就是说, 在 Shuffle 处, 必须断开管道, 进行数据交换, 交换过后, 继续流动, 所以整个流程可以变为如下样子

    

    把 Task 断开成两个部分, Task4 可以从 Task 1, 2, 3 中获取数据, 后 Task4 又作为管道, 继续让数据在其中流动

    但是还有一个问题, 说断开就直接断开吗? 不用打个招呼的呀? 这个断开即没有道理, 也没有规则, 所以可以为这个断开增加一个概念叫做阶段, 按照阶段断开, 阶段的英文叫做 Stage, 如下

    

    所以划分阶段的本身就是设置断开点的规则, 那么该如何划分阶段呢?

    • 第一步, 从最后一个 RDD, 也就是逻辑图中最右边的 RDD 开始, 向前滑动 Stage 的范围, 为 Stage0
    • 第二步, 遇到 ShuffleDependency 断开 Stage, 从下一个 RDD 开始创建新的 Stage, 为 Stage1
    • 第三步, 新的 Stage 按照同样的规则继续滑动, 直到包裹所有的 RDD

    总结来看, 就是针对于宽窄依赖来判断, 一个 Stage 中只有窄依赖, 因为只有窄依赖才能形成数据的 Pipeline.

    如果要进行 Shuffle 的话, 数据是流不过去的, 必须要拷贝和拉取. 所以遇到 RDD 宽依赖的两个 RDD 时, 要切断这两个 RDD 的 Stage.

    这样一个 RDD 依赖的链条, 我们称之为 RDD 的血统, 其中有宽依赖也有窄依赖

• 数据怎么流动？

  // 集群上用 :paste 回车，复制黏贴下面全部代码直接回车 + Ctrl D，它就会执行
  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}")
  
  strRDD.collect.foreach(item => println(item))
  

  上述代码是这个章节我们一直使用的代码流程，如下是其完整的逻辑执行图

  

  • 数据的计算发生在调用 Action 的 RDD 上
  • 如果ResultRDD 没有数据的话，找他的父 RDD 要数据
  • 父RDD再向上要数据

  如果放在集群中运行, 通过 WebUI 可以查看到如下 DAG 结构

  

  Step 1: 从 ResultStage 开始执行最接近 Result 部分的 Stage id 为 0, 这个 Stage 被称之为 ResultStage

  由代码可以知道, 最终调用 Action 促使整个流程执行的是最后一个 RDD, strRDD.collect, 所以当执行 RDD 的计算时候, 先计算的也是这个 RDD

  Step 2: RDD 之间是有关联的前面已经知道, 最后一个 RDD 先得到执行机会, 先从这个 RDD 开始执行, 但是这个 RDD 中有数据吗 ? 如果没有数据, 它的计算是什么? 它的计算是从父 RDD 中获取数据, 并执行传入的算子的函数

  简单来说, 从产生 Result 的地方开始计算, 但是其 RDD 中是没数据的, 所以会找到父 RDD 来要数据, 父 RDD 也没有数据, 继续向上要, 所以, 计算从 Result 处调用, 但是从整个逻辑图中的最左边 RDD 开始, 类似一个递归的过程