经验 - spark中的pipeline机制

今天药忘吃喽~ 2022-03-14 13:48 206阅读 0赞

> ### 问题 ###

如果一个源数据有1亿行, 对这个源数据分别做map()操作和flatMap()操作, 过程是下面描述的那种流程, 为什么?

**1 每读1条数据, 顺次执行map()和flatMap(), 再读取下一条;**

2 对1亿条数据遍历做完map()后, 然后再重新读取一遍这1亿条数据, 做flatMap()操作.

意外的是, 很多人能说出同一个stage的RDD变换, 是一个pipeline操作; 但是对于上面的选择题，基本都表现地非常迟疑, 很少能做出正确的选择, 并说出原因.

上面的答案1是正确的, 原因请见下面的分析.

设有这么一个连续的变换, 会经过sc.textFile() -> map() -> filter() -> mapPartitions() -> flatMap():

sc.txtFile("1.txt").map(_ + 1).filter(_ > 0).mapPartitions{
        iter => {
            dbConnection.open();
            iter.map{
                val result = _ + 2
                if (!iter.hasNext) {
                    dbConnection.close()
                }
                return result
            }
        }
    }.flatMap(_ + 3)

RDD的计算分为2个阶段:

1 通过RDD::compute()方法的嵌套调用, 获得成层叠包含关系的嵌套Iterator的嵌套结构;

2 通过Iterator::next()方法, 从最原始的RDD开始获得数据, 然后通过层叠包含的Iterator对象的层层next()调用, 获取变换后的最终数据;

第1阶段和第二阶段的原理, 分别对应相面2幅图:

![Center][]

图1 通过嵌套调用每个RDD的compute方法, 获得Iterator的包含关系

![Center 1][]

图2 通过调用Iterator的next()的嵌套调用, 获得最终经过一次加工的数据

图中的f()方法, 是用户定义的用于变换(transform)数据的函数字面量; 图2中的mapPartitions()中的f'()方法, 对应用户在函数字面量中, 自行对iterator对象进行包装的函数字面量, 如iter.map()等.

第一阶段:

compute()方法是从最后的RDD开始依次往前调用, RDD4:compute() -> RDD3:compute() -> RDD2:compute() -> RDD1：compute()-> RDD0:compute();

RDD0对应的是HadoopRDD, 其compute()方法是从hdfs文件中读取数据, 返回一个iterator对象;

然后iterator传回给后一个RDD1, RDD1对传入的iterator进行封装产生新的iterator对象, 在next()方法中加入对用户定义的函数字面量的调用以修改(k, v)数据.

就这样, 每个RDD对上一个RDD传入的iterator对象进行封装, 然后将新的iterator对象传给下一个RDD.

第二阶段:

Iterator::next()对象也是从最后一个RDD开始依次往前调用, RDD4:next() -> RDD3:next() -> RDD2:next() -> RDD1:next()-> RDD0:next();

RDD0对应的iterator对象的next(), 真正从hdfs中读取数据, 然后返回(k,v)数据给后一个RDD1;

RDD1及之后的RDD, 都是通过调用用户定义的函数字面量加工数据, 然后返回给后一个RDD.

这样既可看出, 每读一条数据, 都会依次调用各RDD的加工函数, 然后返回给上层应用; 处理完毕后, 在下一个next()方法，又会继续读取下一条数据, 然后循环上面的逻辑.

> ### Spark RDD上的map operators是如何pipeline起来的 ###

最近在工作讨论中，同事提出了这么一个问题：**作用在一个RDD/DataFrame上的连续的多个map是在对数据的一次循环遍历中完成的还是需要多次循环？**

当时我很自然地回答说：不需要多次循环，**spark会将多个map操作pipeline起来apply到rdd partition的每个data element上.**

事后仔细想了想这个问题，虽然我确信spark不可能傻到每个map operator都循环遍历一次数据，但是这些map操作具体是怎么被pipeline起来apply的呢？这个问题还真不太清楚。于是乎，阅读了一些相关源码，力求把这个问题搞清楚。本文就是看完源码后的一次整理，以防过几天又全忘了。

我们从DAGScheduler的submitStage方法开始，分析一下map operators(包括map, filter, flatMap等) 是怎样被pipeline起来执行的。

## submit stage ##

**我们知道，spark的每个job都会被划分成多个stage，这些stage会被DAGScheduler以task set的形式提交给TaskScheduler以调度执行，DAGScheduler的submitStage方法实现了这一步骤。**

![Image 1][]

submitStage in DAGScheduler.scala

如果当前stage没有missingParentStage（未完成的parent stages），submitStage会调用submitMissingTasks，这个方法是做了一些工作的，主要有：

**1. 找到当前stage需要计算的partitions**

stage的partitions就是其对应rdd的partitions，那么stage对应的rdd是怎么确定的呢？源码注释是这样解释的：

***@param rdd RDD that this stage runs on: for a shuffle map stage, it's the RDD we run map tasks on, while for a result stage, it's the target RDD that we ran an action on***

我的理解是：对于shuffle map stage，它的rdd就是引发shuffle的那个operator(比如reduceByKey)所作用的rdd；对于result stage，就是action(比如count)所作用的rdd.

**2. 初始化当前stage的authorizedCommiters**

一个partition对应一个task，当一个task完成后，它会commit它的输出结果到HDFS. 为了防止一个task的多个attempt都commit它们的output，每个task attempt在commit输出结果之前都要向OutputCommitCoordinator请求commit的permission，只有获得批准的attempt才能commit. 批准commit的原则是： "first committer wins" .

在submitMissingTasks方法中会把当前stage的所有partitions对应的tasks的authorizedCommitter都设置为-1，也就是还没有获批的committer.

**3. 获取每个需要计算的partitions的preferred location**

根据每个partition的数据locality信息获取对应task的preferred locations.

**4. 序列化并广播taskBinary**

**taskBinary包含了执行task所需要的信息（包括数据信息，代码信息）。对于不同的task type，taskBinary包含的信息有所不同。spark有两种类型的task ： shuffle map task和result task， 与上面提到的shuffle map stage和result stage相对应。**

shuffle map task的作用是把rdd的数据划分到多个buckets里面，以便shuffle过程使用。这里的划分是依据shuffleDependency中指定的partitioner进行的，所以shuffle map task的taskBinary反序列化后的类型是(RDD\[\_\], ShuffleDependency\[\_, \_, \_\])

result task的作用是在对应的rdd partition上执行指定的function，所以result task的taskBinary反序列化后的类型是(RDD\[T\], (TaskContext, Iterator\[T\]) => U)

生成taskBinary的代码：

![Image 1][]

taskBianry generation in submitMissingTasks of DAGScheduler.scala

**5. 生成tasks**

![Image 1][]

task generation in submitMissingTasks of DAGScheduler.scala

result stage生成result tasks，shuffle map stage生成shuffle map tasks.

有多少个missing partition，就会生成多少个task.

可以看到taskBinary被作为参数用于构建task对象。

**6. 构建task set并向taskScheduler提交**

![Image 1][]

submit task set in submitMissingTasks of DAGScheduler.scala

## spark map operators如何被pipeline的 ##

通过上面的分析，我们知道rdd的map operators最终都会被转化成shuffle map task和result task，然后分配到exectuor端去执行。那么这些map operators是怎么被pipeline起来执行的呢？也就是说shuffle map task和result task是怎么把这些operators串联起来的呢？

为了回答这个问题，我们还需要阅读一下ShuffleMapTask和ResultTask的源码 ：

![Image 1][]

runTask in ShuffleMapTask.scala

![Image 1][]

runTask in ResultTask.scala

shuffle map task和result task都会对taskBinary做反序列化得到rdd对象并且调用rdd.iterator函数去获取对应partition的数据。我们来看看rdd.iterator函数做了什么：

![Image 1][]

iterator in RDD.scala

rdd.iterator调用了rdd.getOrCompute

![Image 1][]

getOrCompute in RDD.scala

getOrCompute会先通过当前executor上的blockManager获取指定block id的block，如果block不存在则调用computeOrReadCheckpoint，computeOrReadCheckpoint会调用compute方法进行计算，而这个compute方法是RDD的一个抽象方法，由RDD的子类实现。

因为filter, map, flatMap操作生成的RDD都是MapPartitionsRDD, 所以我们以MapPartitionsRDD为例：

![Image 1][]

MapPartitionsRDD.scala

可以看到，compute方法调用了parent RDD的iterator方法，然后apply了当前MapPartitionsRDD的f参数. 那这个f又是什么function呢？我们需要回到RDD.scala中看一下map, filter, flatMap的code：

![Image 1][]

map in RDD.scala

![Image 1][]

flatMap in RDD.scala

![Image 1][]

filter in RDD.scala

从上面的源码可以看出，MapPartitionsRDD中的f函数就是对parent rdd的iterator调用了相同的map函数以执行用户给定的function.

所以这是一个逐层嵌套的rdd.iterator方法调用，子rdd调用父rdd的iterator方法并在其结果之上调用scala.collection.Iterator的map函数以执行用户给定的function，逐层调用，直到调用到最初的iterator（比如hadoopRDD partition的iterator）。

**现在，我们最初的问题：“多个连续的spark map operators是如何pipeline起来执行的？” 就转化了“scala.collection.Iterator的多个连续map操作是如何pipeline起来的？”**

## scala.collection.Iterator的map operators是怎么构成pipeline的？ ##

看一下scala.collection.Ierator中map, filter, flatMap函数的源码：

![Image 1][]

map in Iterator.scala

![Image 1][]

filter in Iterator.scala

![Image 1][]

flatMap in Iterator.scala

**从上面的源码可以看出，Iterator的map, filter, flatMap方法返回的Iterator就是基于当前Iterator （self）override了next和hasNext方法的Iterator实例。比如，对于map函数，结果Iterator的hasNext就是直接调用了self iterator的hasNext，next方法就是在self iterator的next方法的结果上调用了指定的map function.**

**flatMap和filter函数稍微复杂些，但本质上一样，都是通过调用self iterator的hasNext和next方法对数据进行遍历和处理。**

**所以，当我们调用最终结果iterator的hasNext和next方法进行遍历时，每遍历一个data element都会逐层调用父层iterator的hasNext和next方法。各层的map function组成了一个pipeline，每个data element都经过这个pipeline的处理得到最终结果数据。**

## 总结 ##

**1. 对RDD的operators最终会转化成shuffle map task和result task在exectuor上执行。**

**2. 每个task （shuffle map task 或 result task）都会被分配一个taskBinary，taskBinary以broadCast的方式分发到每个executor，每个executor都会对taskBinary进行反序列化，得到对应的rdd，以及对应的function或shuffle dependency(function for result task, shuffle dependency for shuffle map task)。**

**3. task通过调用对应rdd的iterator方法获取对应partition的数据，而这个iterator方法又会逐层调用父rdd的iterator方法获取数据。这一过程底层是通过覆写scala.collection.iterator的hasNext和next方法实现的。**

**4. RDD/DataFrame上的连续的map, filter, flatMap函数会自动构成operator pipeline一起对每个data element进行处理，单次循环即可完成多个map operators, 无需多次遍历。**

# #

consult :

[https://www.jianshu.com/p/45c9ee55eea6][https_www.jianshu.com_p_45c9ee55eea6]

[https://blog.csdn.net/cymvp/article/details/54235777][https_blog.csdn.net_cymvp_article_details_54235777]

[Center]: /images/20220313/c4911cddff0f473c95e16e96dd5936ee.png
[Center 1]: /images/20220313/6d62b635f1884ce2b174e21113424355.png
[Image 1]: 
[https_www.jianshu.com_p_45c9ee55eea6]: https://www.jianshu.com/p/45c9ee55eea6
[https_blog.csdn.net_cymvp_article_details_54235777]: https://blog.csdn.net/cymvp/article/details/54235777