【最火大数据 Framework】五分钟深入 Spark 运行机制

上篇文章，我们简要介绍了 MapReduce 框架的局限和 Spark 横空出世的土壤。今天，我们就来详细介绍 Spark 的内部原理和它强大功能的背后设计。

前文回顾

Hadoop 的局限并非只有 IO 速度的问题，更重要的是 MapReduce 规定死了 map / reduce 两种运算，并且提供之间 shuffle 的数据搬运工作。无论运算怎样灵活多样，你都要走 map -> shuffle -> reduce 这条路，要进行灵活运算并保证优秀性能确实有点吃力。Spark 这边，AMPLab 为此做了精心设计，让各种数据处理都能得心应手。

关键概念

Spark 的关键就是引入了 RDD （Resilient Distributed Datasets）的概念。其实没有太深奥，你可以把 RDD 想象成一组数据。

Spark 把要处理的数据，处理中间结果，和输出结果都定义成 RDD. 这样一个常见的 Spark job 就类似于：

• 从数据源读取数据，把输入生成一个 RDD;
• 通过运算把输入 RDD 转换成另一个RDD;
• 再通过运算把生成的 RDD 转换成另一个RDD，重复需要进行的 RDD 转换操作 （此处省略一千遍）;
• 最后运算成结果 RDD，处理结果;

为了处理大量数据，还是把要处理的数据进行分区，分散到多台机器上，以便之后并行处理，这个和 Hadoop 的理念一致。不过，RDD 默认被存到内存中，只有当数据大于 Spark 被允许使用的内存大小时才被 spill 到磁盘（具体内容之后的系列文章会详细介绍）。

RDD 的接口

考虑到 RDD 是连接 Spark 数据操作的核心，RDD 的接口自然是重中之重。简单说，这套接口告诉你：为了生成这个 RDD，它的上一个 RDD 是谁，以及生成过程使用的运算是什么。

抽象？我们换个角度再说一遍。

你有一堆数据 A，你使用了运算 F 把它转换成另一堆数据 B. 那么当我们问，你如何得到 B 时，你怎么回答？我们需要数据 A，并且需要运算 F. 就是这么简单。

在 Spark 里，由于 RDD 被分区存储，所以我们要知道的实际是每个 RDD 分区的来龙去脉。比如：

你有一堆数据 A，被分成了 A1，A2 两个分区，你为每个分区使用了运算 F 把它们转换成另一堆数据 B1，B2，合起来就是B。那么当我们问，你如何得到 B2 时，你怎么回答？我们需要数据 A2，并且需要运算 F. 同样的，你如何得到 B1 ？我们需要数据 A1, 并且需要运算 F. 就是这么简单。

数据操作

别小看了 RDD 的这套接口，掐指一算，绝大多数运算都可以由这套接口定义。我们看几个例子：

Map

其实我们上面的例子就是 map：一个 RDD 的分区分别转换成下一个 RDD 的分区，各个分区之间互不影响。那每个 RDD 分区的 “爸爸“ 就是上一个 RDD 对应的分区。运算就是用户定义的map function.

Filter

还是用上面的例子：只不过这个 F 变成“这条数据是否该留下来”，在这种情况下这样 A1 >= B1.

GroupbyKey

这个复杂一些，它里面的数据不是单个的，而是 key-value pair. 联系我们之前 Hadoop 的例子，RDD B 里的分区中的数据有可能是 A1，也有可能是A2 里的，那我们就清清楚楚地告诉 B，你的每个分区的 “爸爸” 都是 A 里面所有的分区。运算呢？就是合并所有 Key 一样的 key value pair，组成一个 set.

ReduceByKey ®

这个比 groupBy 再复杂一点。B 里的每个分区的 ”爸爸“ 还是 A 里面所有的分区。运算呢？就是合并所有 Key 一样的 key value pair, 然后为所有同样的Key 运行 R 这个function.

Spark RDD 支持的运算很多很多，但是本质都是用 RDD 的接口灵活的定义出不同运算。用户也可以根据自己需要创作新的运算。这样 Spark 允许用户用不同种类的运算实现了复杂的企业逻辑，甚至是 SQL 的处理和机器学习。这看似简单的设计恰恰是 Spark 强大的基础。

RDD的构建

从上面的学习我们可以发现 RDD 其实就是数据集，是一组数据被处理到一个阶段的状态。

每一个 Spark Job 就是定义了由输入 RDD，如何把它转化成下一个状态，再下一个状态 …… 直到转化成我们的输出。这些转化就是对 RDD 里每一个 data record 的操作。用个高大上点的语言，一个 Spark job 就是一系列的 RDD 以及他们之间的转换关系。那么用户如何才能定义 RDD 和转换关系呢？换句话说，用户如何使用 Spark 呢？

可能绝大多数读者没有学习过 Scala，那么我们就用大家更熟悉的 Java 语言描述，Spark 也提供了Java语言的支持。

用户需要定义一个包含主函数的 Java (main) 类。在这个 main 函数中，无论业务逻辑多么复杂，无论你需要使用多少 Java 类，如果从 Spark 的角度简化你的程序，那么其实就是：

• 首先生成 JavaSparkContext 类的对象.
• 从 JavaSparkContext 类的对象里产生第一个输入RDD. 以读取 HDFS 作为数据源为例，调用 JavaSparkContext.textFile() 就生成第一个 RDD.
• 每个 RDD 都定义了一些标准的常用的变化，比如我们上面提到的 map, filter, reduceByKey …… 这些变化在 Spark 里叫做 transformation.
• 之后可以按照业务逻辑，调用这些函数。这些函数返回的也是 RDD, 然后继续调用，产生新的RDD …… 循环往复，构建你的 RDD 关系图。
• 注意 RDD 还定义了其他一些函数，比如 collect, count, saveAsTextFile 等等，他们的返回值不是 RDD. 这些函数在 Spark 里叫做 actions, 他们通常作为 job 的结尾处理。
• 用户调用 actions 产生输出结果，Job 结束。

补充说明 action

Action 都是类似于 “数数这个 RDD 里有几个 data record”, 或者 ”把这个 RDD 存入一个文件” 等等。想想他们作为结尾其实非常合理：我们使用 Spark 总是来实现业务逻辑的吧？处理得出的结果自然需要写入文件，或者存入数据库，或者数数有多少元素，或者其他一些统计什么的。所以 Spark 要求只有用户使用了一个 action，一个 job 才算结束。当然，一个 job 可以有多个 action，比如我们的数据既要存入文件，我们又期望知道有多少个元素。

这些 RDD 组成的关系在 Spark 里叫做 DAG，就是有向无循环图，图论里的一个概念，大家有兴趣可以专门翻翻这个概念。可以发现，实践中绝大部分业务逻辑都可以用 DAG 表示，所以 spark 把 job 定义成 DAG 也就不足为奇了。

RDD 的两种变化

我们上面刚刚介绍了 transformation 的概念。在 Spark 眼中，transformation 被分成 narrow transformation 和 wide transformation. 这是什么东西呢？

上文提到过 RDD 被分成几个分区，分散在多台机器上。当我们把一个 RDD A 转化成下一个 RDD B 时，这里有两种情况：

1. 有时候只需要一个 A 里面的一个分区，就可以产生 B 里的一个分区了，比如 map 的例子：A 和 B 之间每个分区是一一对应的关系，这就是 narrow transofmration.
2. 还有一类 transformation，可能需要 A 里面所有的分区，才能产生 B 里的一个分区，比如 reduceByKey的例子，这就是 wide transformation.

Narrow 或者 Wide 有什么关系吗？

一个 Spark job 中可能需要连续地调用 transformation, 比如先 map，后 filter，然后再 map …… 那这些 RDD 的变化用图表示就是：

我们可以大胆设想一下，如果每个分区里的数据就待在那台机器的内存里，我们逐一的调用 map, filter, map 函数到这些分区里，Job 就很好的完成。

更重要的是，由于数据没有转移到别的机器，我们避免了 Network IO 或者 Disk IO. 唯一的任务就是把 map / filter 的运行环境搬到这些机器上运行，这对现代计算机来说，overhead 几乎可以忽略不计。

这种把多个操作合并到一起，在数据上一口气运行的方法在 Spark 里叫 pipeline （其实 pipeline 被广泛应用的很多领域，比如 CPU）。这时候不同就出现了：只有 narrow transformation 才可以进行 pipleline 操作。对于 wide transformation, RDD 转换需要很多分区运算，包括数据在机器间搬动，所以失去了 pipeline 的前提。

RDD 的执行

当用户调用 actions 函数时，Spark 会在后台创建出一个 DAG. 就是说 Spark 不仅用 DAG 建模，而且真正地创建出一个 DAG, 然后执行它（顺便说一句 DAG 在 Spark 里不是用一个对象表示的，而是用 RDD 对象之间的关系，之后系列文章会深入学习）。

Spark 会把这个 DAG 交给一个叫 DAG scheduler 的模块，DAG scheduler 会优先使用 pipeline 方法，把 RDD 的 transformation 压缩；当我们遇到 wide transformation 时，由于之前的 narrow transformation 无法和 wide transformation pipeline, 那 DAG scheduler 会把前面的 transformation 定义成一个 stage.

重要的事情说三遍：DAG scheduler 会分析 Spark Job 所有的 transformation, 用 wide transformation 作为边界，把所有 transformation 分成若干个stages. 一个 stage 里的一个分区就被 Spark 叫做一个task. 所以一个 task 是一个分区的数据和数据上面的操作，这些操作可能包括一个 transformation，也可能是多个，但一定是 narrow transformation.

DAG scheduler 工作的结果就是产生一组 stages. 这组 stages 被传到 Spark 的另一个组件 task scheduler, task scheduler 会使用集群管理器依次执行 task, 当所有的 task 执行完毕，一个 stage 标记完成；再运行下一个 stage …… 直到整个 Spark job 完成。

结语

Spark 能提供强大的功能和广泛的支持性，奥妙就在于 RDD. 整个思路看起来很简单，就用它把 distributed computation 推到前所未有的高度；你也可以说它很深奥，为什么要如此定义分布式运算？为什么要把运算分成 narrow transformation 和 wide transformation？希望本文对大家有所帮助。之后我们会继续深入学习 Spark 的方方面面，通过探讨 Spark 的众多细节来掌握它的运算模型的精髓。欢迎大家持续关注。