【Spark四十六】Spark RDD物理执行图与Stage

理解记忆如下要点：

1. 划分Stage的方法是从后向前，遇到 shuffle Dependency就断开，如果是Narrow Dependency就划入当前的Stage。所以，ShuffledRDD通常是一个Stage的开头(Stage的开头未必是 ShuffledRDD，也有可能是数据源转换而来的RDD)。所以上面一共3个Stage。需要注意的是，shuffle的结果未必成为 ShuffledRDD，例如上面的CoGroupedRDD，一半来源于Shuffle。

2. 每个 stage 里面 task 的数目由该 stage 最后一个 RDD 中的 partition 个数决定。Task分为两类，一类是ShuffleMapTask，一类是ResultTask。而且一个Stage中的Task类型只能是其中一种类型

3. 数据用的时候再算，而且数据是流到要计算的位置的。所谓用到的时候再算，是指从后向前，本RDD追溯父RDD的数据，直到数据源，然后数据源的原始数据，依次经历所有的RDD的算子转换操作。所谓的计算位置，指的是ShuffleMapTask或者ResultTask运行的地方

4.对于没有 parent stage 的 stage，该 stage 最左边的 RDD 是可以立即计算的，而且每计算出一个 record 后便可以流入 f 或 g（所谓的f，g是指RDD链上的不同RDD的算子操作）。如 果 f 中的 record 关系是 1:1 的，那么 f(record1) 计算结果可以立即顺着 computing chain 流入 g 中。如果 f 的 record 关系是 N:1，record1 进入 f() 后也可以被回收。总结一下，computing chain 从后到前建立，而实际计算出的数据从前到后流动，而且计算出的第一个 record 流动到不能再流动后，再计算下一个 record。这样，虽然是要计算后续 RDD 的 partition 中的 records，但并不是要求当前 RDD 的 partition 中所有 records 计算得到后再整体向后流动。(这段话描述了，RDD的元素计算不是全部加载然后再计算，是边加载边计算的模式)

5. 对于有 parent stage 的 stage，先等着所有 parent stages 中 final RDD 中数据计算好，然后经过 shuffle 后，问题就又回到了计算 “没有 parent stage 的 stage”。这里明确了Stage依赖之后，依赖的Stage的任务必须先执行完，然后才开始下一个任务。而不是依赖的Stage任务完成了一部分，然后下个Stage的任务就开始执行了。

6. RDD的实现细节：

6.1每个 RDD 包含的 getDependency() 负责确立 RDD 的数据依赖

6.2 compute() 方法负责接收 parent RDDs 或者 data block 流入的 records，进行计算，然后输出 record。

经常可以在 RDD 中看到这样的代码firstParent[T].iterator(split, context).map(f)。firstParent 表示该 RDD 依赖的第一个 parent RDD，iterator() 表示 parentRDD 中的 records 是一个一个流入该 RDD 的，map(f) 表示每流入一个 recod 就对其进行 f(record) 操作，输出 record。为了统一接口，这段 compute() 仍然返回一个 iterator，来迭代 map(f) 输出的 records。

总结一下：整个 computing chain 根据数据依赖关系自后向前建立，遇到 ShuffleDependency 后形成 stage。在每个 stage 中，每个 RDD 中的 compute() 调用 parentRDD.iter() 来将 parent RDDs 中的 records 一个个 fetch 过来。

 

 

给定一个逻辑执行图，如何进行Stage划分，如下所示的逻辑执行图，包含几个Stage？

 

 

 

 

 cartesian算子对应的窄依赖逻辑图和物理图

 

 上图中，包含了六个ResultTask，仅有ResultTask，比如黑色线对应的这个Result，它需要计算两个数据块，然后对3个RDD的partition进行处理，这一整套的过程都是在一个ResultTask中执行的，ResultTask个数与最后一个RDD的分区数(这里是CartesianRDD)相同

 

插播一个computing chain的图片，在最开始出用到了