专题介绍

Spark于2009年诞生于加州大学伯克利分校的算法(机器与人实验室)，2010年开源。Spark于2013年捐赠给ApacheSoftwareFoundation，2014年成为Apache的顶级项目。

如今，十年过去了，Spark已经成为大大小小的企业和研究机构的常用工具之一，依然深受众多开发者的喜爱。如果你初入江湖，想了解和学习星火的& quot小虾米& quot，那么InfoQ和飞轮技术专家Leo的系列文章—— 《深入浅出Spark：原理详解与开发实践》一定适合你！

本文是专题系列的第二篇。

这本书上接前一篇文章。在上一篇文章《内存计算的由来——RDD》中，我们从& quot虚拟的和& quot真实& quot。RDD是通过端到端连接依赖关系和计算属性形成的计算路径。用专业术语来说，叫做Lineage—— lineage，也叫DAG(DirectedAcyclicGraph)。为什么一个概念有两个名字？这两个不同的名字有什么区别和联系？简单来说，lineage和DAG从两个不同的角度描述了同一个东西。谱系，侧重于从数据的角度描述不同rdd之间的依赖关系；而DAG则是从计算的角度来描述不同rdd之间的转换逻辑。如果说RDD是Spark对分布式数据模型的抽象，那么DAG就是Spark对分布式计算模型的抽象。

顾名思义，DAG是一种图。图计算模型的使用由来已久，早在上个世纪就被应用于图数据库的实现。任何图都包含两个基本元素：顶点和边。节点通常用于表示实体，而边表示实体之间的关系。例如，在社交网络的朋友关系中& quot永恒屠龙者& quot，每个节点代表一个特定的人，每条边意味着两端的实体之间已经建立了朋友关系。

永恒屠龙者社交网络

在上述社交网络中，朋友之间的关系是相互的。比如张无忌和周芷若是朋友，那么图中的边是没有方向性的。另外，细心的同学可能已经发现，还有& quot戒指& quot在上图结构中，如张无忌、谢逊与白眉鹰王形成的关系环，张无忌、谢逊、紫龙王与小昭的关系环，等等。像上面这样的图形结构叫做& quot无向循环图& quot。没有比较，就没有歧视。有向无环图(DAG)自然是一种具有方向性和无& quot戒指& quot结构。各位读者还记得土豆工坊的例子吗？

马铃薯车间山

在上面的马铃薯加工DAG中，每个节点是一个RDD，每个边代表不同rdd之间的父子关系。父子关系自然是单向的，所以整个画面是有方向性的。此外，我们注意到整张图片中没有环形结构。像这样的土豆加工流水线，可以说是最简单的有向无环图。每个节点的入度和出度都是1，整个图只有一个分支。

但工业应用中的SparkDAG要比这复杂得多，它往往是一个有向无环图，由不同rdd的关联和拆分产生多个分支。为了说明这一点，我们以土豆工坊为例。把& quot原创& quot薯片上市一段时间，作坊老板发现季度销量暴跌，老板又担心又无奈。这时，有人向他建议：& quot为什么不推出更多口味的薯片来迎合大众多样化的选择& quot，于是老板下了一道命令，工人们对装配线做了如下改动。

马铃薯车间先进生产线

与以前相比，新工艺增加了三条调味生产线，用于调配不同的调味粉。新生产线上的辣椒粉分配到收集小薯片的生产线，孜然粉分配到中薯片生产线，番茄粉相应分配到大薯片生产线。改造后的土豆车间现在可以生产三种口味不同大小的薯片，分别是麻辣口味的小薯片、孜然口味的中薯片和番茄口味的大薯片。如果我们使用flavoursRDD来抽象调味品，那么对应于新车间工作流的DAG将演变成一个有向无环图，有两个分支，如下所示。

多分支DAG

在上一篇文章中，我们讨论了SparkCore内部心法的第一个本质，——RDD。在本文中，我们来谈谈内心法的第二个秘密，——DAG。

RDD算子——DAG的边

在上一篇文章《内存计算的由来——RDD》的最后，我们以WordCount为例，展示了不同rdd之间转换形成的DAG计算图。通读代码，从开发的角度，我们发现DAG的关键是RDD算子调用。与HadoopMapReduce不同，Spark提供了丰富的基于数据的RDD算子，允许开发者灵活排列组合，从而实现多样化的数据处理逻辑。那么问题来了，Spark提供了哪些算子？

资料来源：https://spark.apache.org/docs/latest/rdd-programming-guide.html.

从表中，我们可以看到Spark的RDD运营商令人眼花缭乱。对于刚接触Spark的同学来说，如果不稍微分类的话，真的没办法从这么多运营商入手。ApacheSpark官网将RDD算子分为变换和动作，这在各种Spark技术博客中也很常见。

的分类方法。为了说明Transformations和Actions算子的本质区别，我们必须得提一提Spark计算模型的“惰性计算”（Lazyevaluation，又名延迟计算）特性。

掌握一个新概念最有效的方法之一就是找到与之相对的概念——与“惰性计算”相对，大多数传统编程语言、编程框架的求值策略是“及早求值”（Eagerevaluation）。例如，对于我们熟悉的C、C++、Java来说，每一条指令都会尝试调度CPU、占用时钟周期、触发计算的执行，同时，CPU寄存器需要与内存通信从而完成数据交换、数据缓存。在传统编程模式中，每一条指令都很“急”（Eager），都恨不得自己马上被调度到“前线”、参与战斗。

惰性计算模型则不然——具体到Spark，绝大多数RDD算子都很“稳”、特别能沉得住气，他们会明确告诉DAGScheduler：“老兄，你先往前走着，不用理我，我先绷会儿、抽袋烟。队伍的前排是我们带头大哥，没有他的命令，我们不会贸然行动。”有了惰性计算和及早求值的基本了解，我们再说回Transformations和Actions的区别。在Spark的RDD算子中，Transformations算子都属于惰性求值操作，仅参与DAG计算图的构建、指明计算逻辑，并不会被立即调度、执行。惰性求值的特点是当且仅当数据需要被物化（Materialized）时才会触发计算的执行，RDD的Actions算子提供各种数据物化操作，其主要职责在于触发整个DAG计算链条的执行。当且仅当Actions算子触发计算时，DAG从头至尾的所有算子（前面用于构建DAG的Transformations算子）才会按照依赖关系的先后顺序依次被调度、执行。

说到这里，各位看官不禁要问：Spark采用惰性求值的计算模型，有什么优势吗？或者反过来问：Spark为什么没有采用传统的及早求值？不知道各位看官有没有听说过“延迟满足效应”（又名“糖果效应”），它指的是为了获取长远的、更大的利益而自愿延缓甚至放弃目前的、较小的满足。正所谓：“云想衣裳花想容，猪想发福人想红”。Spark这孩子不仅天资过人，小小年纪竟颇具城府，独创的内功心法意不在赢得眼下的一招半式，而是着眼于整个武林。扯远了，我们收回来。笼统地说，惰性计算为Spark执行引擎的整体优化提供了广阔的空间。关于惰性计算具体如何帮助Spark做全局优化——说书的一张嘴表不了两家事，后文书咱们慢慢展开。

还是说回RDD算子，除了常见的按照Transformations和Actions分类的方法，笔者又从适用范围和用途两个维度为老铁们做了归类，毕竟人类的大脑喜欢结构化的知识，官网上一字长蛇阵的罗列总是让人看了昏昏欲睡。有了这个表格，我们就知道*ByKey的操作一定是作用在PairedRDD上的，所谓PairedRDD是指Schema明确区分（Key,Value）对的RDD，与之相对，任意RDD指的是不带Schema或带任意Schema的RDD。从用途的角度来区分RDD算子的归类相对比较分散，篇幅的原因，这里就不一一展开介绍，老铁们各取所需吧。

值得一提的是，对于相同的计算场景，采用不同算子实现带来的执行性能可能会有天壤之别，在后续的性能调优篇咱们再具体问题具体分析。好吧，坑越挖越多，列位看官您稍安勿躁，咱们按照FIFO的原则，先来说说刚刚才提到的、还热乎的DAGScheduler。

DAGScheduler——DAG的向导官

DAGScheduler是Spark分布式调度系统的重要组件之一，其他组件还包括TaskScheduler、MapOutputTracker、SchedulerBackend等。DAGScheduler的主要职责是根据RDD依赖关系将DAG划分为Stages，以Stage为粒度提交任务（TaskSet）并跟踪任务进展。如果把DAG看作是Spark作业的执行路径或“战略地形”，那么DAGScheduler就是这块地形的向导官，这个向导官负责从头至尾将地形摸清楚，根据地形特点排兵布阵。更形象地，回到土豆工坊的例子，DAGScheduler要做的事情是把抽象的土豆加工DAG转化为工坊流水线上一个个具体的薯片加工操作任务。那么问题来了，DAGScheduler以怎样的方式摸索“地形”？如何划分Stages？划分Stages的依据是什么？更进一步，将DAG划分为Stages的收益有哪些？Spark为什么要这么做？

DAGScheduler的核心职责

为了回答这些问题，我们需要先对于DAG的“首”和“尾”进行如下定义：在一个DAG中，没有父RDD的节点称为首节点，而没有子RDD的节点称为尾节点。还是以土豆工坊为例，其中首节点有两个，分别是potatosRDD和flavoursRDD，而尾节点是flavouredBakedChipsRDD。

DAG中首与尾的定义

DAGScheduler在尝试探索DAG“地形”时，是以首尾倒置的方式从后向前进行。具体说来，对于土豆工坊的DAG，DAGScheduler会从尾节点flavouredBakedChipsRDD开始，根据RDD依赖关系依次向前遍历所有父RDD节点，在遍历的过程中以Shuffle为边界划分Stage。Shuffle的字面意思是“洗牌”，没错，就是扑克游戏中的洗牌，在大数据领域Shuffle引申为“跨节点的数据分发”，指的是为了实现某些计算逻辑需要将数据在集群范围内的不同计算节点之间定向分发。在绝大多数场景中，Shuffle都是当之无愧的“性能瓶颈担当”，毫不客气地说，有Shuffle的地方，就有性能优化的空间。关于SparkShuffle的原理和性能优化技巧，后面我们会单独开一篇来专门探讨。在土豆工坊的DAG中，有两个地方发生了Shuffle，一个是从bakedChipsRDD到flavouredBakedChipsRDD的计算，另一个是从flavoursRDD到flavouredBakedChipsRDD的计算，如下图所示。

土豆工坊DAG中的Shuffle

各位看官不禁要问：DAGScheduler如何判断RDD之间的转换是否会发生Shuffle呢？那位看官说了：“前文书说了半天算子是RDD之间转换的关键，莫不是根据算子来判断会不会发生Shuffle？”您还真猜错了，算子与Shuffle没有对应关系。就拿join算子来说，在大部分场景下，join都会引入Shuffle；然而在collocatedjoin中，左右表数据分布一致的情况下，是不会发生Shuffle的。所以您看，DAGScheduler还真不能依赖算子本身来判断发生Shuffle与否。要回答这个问题，咱们还是得回到前文书《内存计算的由来——RDD》中介绍RDD时提到的5大属性。

属性名成员类型属性含义dependencies变量生成该RDD所依赖的父RDDcompute方法生成该RDD的计算接口partitions变量该RDD的所有数据分片实体partitioner方法划分数据分片的规则preferredLocations变量数据分片的物理位置偏好

RDD的5大属性及其含义

其中第一大属性dependencies又可以细分为NarrowDependency和ShuffleDependency，NarrowDependency又名“窄依赖”，它表示RDD所依赖的数据无需分发，基于当前现有的数据分片执行compute属性封装的函数即可；ShuffleDependency则不然，它表示RDD依赖的数据分片需要先在集群内分发，然后才能执行RDD的compute函数完成计算。因此，RDD之间的转换是否发生Shuffle，取决于子RDD的依赖类型，如果依赖类型为ShuffleDependency，那么DAGScheduler判定：二者的转换会引入Shuffle。在回溯DAG的过程中，一旦DAGScheduler发现RDD的依赖类型为ShuffleDependency，便依序执行如下3项操作：

沿着Shuffle边界的子RDD方向创建新的Stage对象把新建的Stage注册到DAGScheduler的stages系列字典中，这些字典用于存储、记录与Stage有关的状态和元信息，以备后用沿着当前RDD的父RDD遵循广度优先搜索算法继续回溯DAG

拿土豆工坊来说，其尾节点flavouredBakedChipsRDD同时依赖bakedChipsRDD和flavoursRDD两个父RDD，且依赖类型都是ShuffleDependency，那么依据DAGScheduler的执行逻辑，此时会执行如下3项具体操作：

DAGScheduler回溯DAG过程当中遇到ShuffleDependency时的主要操作流程

DAGScheduler沿着尾节点回溯并划分出stage0

在完成第一个Stage（stage0）的创建和注册之后，DAGScheduler先沿着bakedChipsRDD方向继续向前回溯。在沿着这条路向前跑的时候，我们的这位DAGScheduler向导官惊喜地发现：“我去！这一路上一马平川、风景甚好，各个驿站之间什么障碍都没有，交通甚是顺畅，真是片好地形！”——沿路遇到的所有RDD（bakedChipsRDD，chipsRDD，cleanedPotatosRDD，potatosRDD）的依赖类型都是NarrowDependency。

在回溯完毕时，DAGScheduler同样会重复上述3个步骤，根据DAGScheduler以Shuffle为边界划分Stage的原则，沿途的所有RDD都划归为同一个Stage，暂且记为stage1。值得一提的是，Stage对象的rdd属性对应的数据类型是RDD[]，而不是List[RDD[]]。对于一个逻辑上包含多个RDD的Stage来说，其rdd属性存储的是路径末尾的RDD节点，具体到我们的案例中就是bakedChipsRDD。

DAGScheduler沿着bakedChipsRDD方向回溯并划分出stage1

勤勤恳恳的DAGScheduler在成功创建stage1之后，依然不忘初心、牢记使命，继续奔向还未探索的路线。从上图中我们清楚地看到整块地形还剩下flavoursRDD方向的路径没有纳入DAGScheduler的视野范围。咱们的这位DAGScheduler向导官记性相当得好，早在划分stage0的时候，他就用小本子（栈）记下：“此路口有分叉，先沿着bakedChipsRDD方向走，然后再回过头来沿着flavoursRDD的方向探索。切记，切记！”此时，向导大人拿出之前的小本子，用横线把bakedChipsRDD方向的路径划掉——表示该方向路径已探索过，然后沿着flavoursRDD方向大踏步地走下去。一脚下去，发现：“我去！到头儿了！”，然后紧接着执行一贯的“三招一套”流程——创建Stage、注册Stage、继续回溯。随着DAGScheduler创建最后一个Stage：stage2，地形上的所有路径都已探索完毕。

DAGScheduler创建最后一个Stage：stage2

到此为止，我们的向导大人几乎跑断了腿、以首尾倒置的顺序对整片地形进行了地毯式搜查，最终将地形划分为3块战略区域（Stage）。那么问题来了，向导大人划分出的3块区域，有啥用呢？DAGScheduler他老人家马不停蹄地这么跑，到底图啥？前面我们提到，DAGScheduler的核心职责，是将抽象的DAG计算图转换为具体的、可并行计算的分布式任务。回溯DAG、创建Stage，只是这个核心职责的第一步，DAGScheduler以Stage（TaskSet）为粒度进行任务调度，伙同TaskScheduler、SchedulerBackend等一众大佬运筹帷幄、调兵遣将。不过，毕竟本篇的主题是DAG，到Spark调度系统的核心还有些距离，因此这里咱们暂且挖个坑，后面再单独开篇（Spark调度系统）专门讲述几位大佬之间的趣事逸闻。填坑之路漫漫其修远兮，吾将上下而挖坑。

咱们来回顾一下向导大人的心路历程，首先，DAGScheduler沿着DAG的尾节点一路北上，并沿途判断每一个RDD节点的dependencies属性。之后，如果判定RDD的dependencies属性是NarrowDependency，则DAGScheduler继续向前回溯；若RDD的依赖是ShuffleDependency，DAGScheduler便开启“三招一套”的招式，创建Stage、注册Stage并继续向前回溯。由此可见，何时切割DAG并生成新的Stage由RDD的依赖类型决定，当且仅当RDD的依赖是ShuffleDependency时，DAGScheduler才会新建Stage。

喜欢刨根问底的您一定会问：“DAGScheduler怎么知道RDD的依赖类型到底是哪一个？他怎么判别RDD的依赖是窄依赖还是ShuffleDependency？”要回答这个问题，我们就还得回到RDD的5大属性上，不过这次出场的是partitioner。还记得这个属性吗？partitioner是RDD的分区器、定义了RDD数据分片的分区规则，它决定了RDD的数据分片在分布式集群中如何分布，这个属性至关重要，后面介绍Shuffle的时候我们还会提到它。DAGScheduler正是通过partitioner来判定每个RDD的依赖类型，具体来说，如果子RDD的partitioner与父RDD的partitioner一致，那么DAGScheduler判定子RDD对父RDD的依赖属于窄依赖；相反，如果两者partitioner不一致，也即分区规则不同（分区规则不同则意味着一定存在数据的“重洗牌”，即Shuffle），那么DAGScheduler判定子对父的依赖关系是ShuffleDependency。到此，DAGScheduler对于DAG的划分逻辑可以暂且告一段落。原理说了，例子举了，还缺啥？对！代码。

Showmethecode

古人云：“光说不练假把式”，我们用一个小例子来展示一下DAG与Stage的关系。还是用上篇《内存计算的由来——RDD》中的WordCount依样画葫芦，文件内容如下。

示例文件内容

代码也没变：

WordCount示例代码

虽然文件内容和代码都没变，但是我们观察问题的视角变了，这次我们关心的是DAG中Stage的划分以及Stage之间的关系。RDD的toDebugString函数让我们可以一览DAG的构成以及Stage的划分，如下图所示。

DAG构成及Stage划分

在上图中，从第3行往下，每一行表示一个RDD，很显然，第3行的ShuffledRDD是DAG的尾节点，而第7行的HadoopRDD是首节点。我们来观察每一行字符串打印的特点，首先最明显地，第4、5、6、7行的前面都有个制表符（Tab），与第3行有个明显的错位，这表示第3行的ShuffledRDD被划分到了一个Stage（记为stage0），而第4、5、6、7行的其他RDD被划分到了另外一个Stage（记为stage1），且stage0对stage1有依赖关系。假设第7行下面的RDD字符串打印有两个制表符，即与第7行产生错位，那么第7行下面的RDD则被划到了新的Stage，以此类推。

由此可见，通过RDD的toDebugString观察DAG的Stage划分时，制表符是个重要的指示牌。另外，我们看到第3、4行的开头都有个括号，括号里面是数字，这个数字标记的是RDD的partitions大小。当然了，观察RDD、DAG、Stage还有更直观的方式，Spark的WebUI提供了更加丰富的可视化信息，不过Spark的WebUI面板繁多，对于新同学来说一眼望去反而容易不知所措，也许后面时间允许的话我们单开一篇SparkWebUI的串讲。

Postscript

本篇是《Spark分布式计算科普专栏》的第二篇，笔者学浅才疏、疏漏难免。如果您有任何疑问，或是觉得文章中的描述有所遗漏或不妥，欢迎在评论区留言、讨论。掌握一门技术，书本中的知识往往只占两成，三成靠讨论，五成靠实践。更多的讨论能激发更多的观点、视角与洞察，也只有这样，对于一门技术的认知与理解才能更深入、牢固。

在本篇博文中，我们从DAG的边——SparkRDD算子入手，介绍了衔接RDD的两大类算子：Transformations和Actions，并对惰性计算有了初步的认知。然后，还是以土豆工坊为例，介绍DAGScheduler切割DAG、生成Stage的流程和步骤，尤其需要注意的是DAGScheduler以Shuffle为边界划分Stage。

最后，用上一篇的WordCount简单展示了DAG与Stage的关系。细心的读者可能早已发现，文中多次提及“后文书再展开”、“后面再单开一篇”，Spark是一个精妙而复杂的分布式计算引擎，在本篇博文中我们不得不对Spark中的许多概念都进行了“前置引用”。换句话说，有些概念还没来得及解释（如惰性计算、Shuffle、TaskScheduler、TaskSet、Spark调度系统），就已经被引入到了本篇博文中。这样的叙述方法也许会给您带来困惑，毕竟，用一个还未说清楚的概念，去解释另一个新概念，总是感觉没那么牢靠。

常言道：“杀人偿命、欠债还钱”，在后续的专栏文章中，我们会继续对Spark的核心概念与原理进行探讨，慢慢地把欠您的技术债还上，尽可能地还原Spark分布式内存计算引擎的全貌。毕竟Spark调度系统为何方神圣，DAGScheduler伙同TaskScheduler、SchedulerBackend、TaskSetManager等一众大佬如何演绎权利的游戏，且听下回分解。

作者简介

吴磊，SparkSummitChina2017讲师、WorldAIConference2020讲师，曾任职于IBM、联想研究院、新浪微博，具备丰富的数据库、数据仓库、大数据开发与调优经验，主导基于海量数据的大规模机器学习框架的设计与实现。现担任ComcastFreewheel机器学习团队负责人，负责计算广告业务中机器学习应用的实践、落地与推广。热爱技术分享，热衷于从生活的视角解读技术，曾于《IBMdeveloperWorks》和《程序员》杂志发表多篇技术文章。

延伸阅读：

深入浅出Spark（一）：内存计算的由来-InfoQ

关注我并转发此篇文章，私信我“领取资料”，即可免费获得InfoQ价值4999元迷你书，点击文末「了解更多」，即可移步InfoQ官网，获取最新资讯～