Apache Spark 的设计与实现(整体介绍)

概览

拿到系统后，部署系统是第一件事，那么系统部署成功之后， 各个节点都启动了哪些服务？

部署图

从部署图中能够看到

• 整个集群分为 Master 节点和 Worker 节点，至关于 Hadoop 的 Master 和 Slave 节点。
• Master 节点上常驻 Master 守护进程，负责管理所有的 Worker 节点。
• Worker 节点上常驻 Worker 守护进程，负责与 Master 节点通讯并管理 executors。
• Driver 官方解释是 “The process running the main() function of the application and creating the SparkContext”。Application 就是用户本身写的 Spark 程序（driver program），好比 WordCount.scala。若是 driver program 在 Master 上运行，好比在 Master 上运行

  ./bin/run-example SparkPi 10

    那么 SparkPi 就是 Master 上的 Driver。若是是 YARN 集群，那么 Driver 可能被调度到 Worker 节点上运行（好比上图中的 Worker Node 2）。另外，若是直接在本身的 PC 上运行 driver program，好比在 Eclipse 中运行 driver program，使用

  val sc = new SparkContext("spark://master:7077", "AppName")

    去链接 master 的话，driver 就在本身的 PC 上，可是不推荐这样的方式，由于 PC 和 Workers 可能不在一个局域网，driver 和 executor 之间的通讯会很慢。

• 每一个 Worker 上存在一个或者多个 ExecutorBackend 进程。每一个进程包含一个 Executor 对象，该对象持有一个线程池，每一个线程能够执行一个 task。
• 每一个 application 包含一个 driver 和多个 executors，每一个 executor 里面运行的 tasks 都属于同一个 application。
• 在 Standalone 版本中，ExecutorBackend 被实例化成 CoarseGrainedExecutorBackend 进程。

  在我部署的集群中每一个 Worker 只运行了一个 CoarseGrainedExecutorBackend 进程，没有发现如何配置多个 CoarseGrainedExecutorBackend 进程。（应该是运行多个 applications 的时候会产生多个进程，这个我尚未实验，） 想了解 Worker 和 Executor 的关系详情，能够参阅 @OopsOutOfMemory 同窗写的 Spark Executor Driver资源调度小结 。

• Worker 经过持有 ExecutorRunner 对象来控制 CoarseGrainedExecutorBackend 的启停。

了解了部署图以后，咱们先给出一个 job 的例子，而后概览一下 job 如何生成与运行。

Job 例子

咱们使用 Spark 自带的 examples 包中的 GroupByTest，假设在 Master 节点运行，命令是

/* Usage: GroupByTest [numMappers] [numKVPairs] [valSize] [numReducers] */

bin/run-example GroupByTest 100 10000 1000 36

  GroupByTest 具体代码以下

package org.apache.spark.examples

import java.util.Random

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.SparkContext._

/**
  * Usage: GroupByTest [numMappers] [numKVPairs] [valSize] [numReducers]
  */
object GroupByTest {
  def main(args: Array[String]) {
    val sparkConf = new SparkConf().setAppName("GroupBy Test")
    var numMappers = 100
    var numKVPairs = 10000
    var valSize = 1000
    var numReducers = 36

    val sc = new SparkContext(sparkConf)

    val pairs1 = sc.parallelize(0 until numMappers, numMappers).flatMap { p =>
      val ranGen = new Random
      var arr1 = new Array[(Int, Array[Byte])](numKVPairs)
      for (i <- 0 until numKVPairs) {
        val byteArr = new Array[Byte](valSize)
        ranGen.nextBytes(byteArr)
        arr1(i) = (ranGen.nextInt(Int.MaxValue), byteArr)
      }
      arr1
    }.cache
    // Enforce that everything has been calculated and in cache
    pairs1.count

    println(pairs1.groupByKey(numReducers).count)

    sc.stop()
  }
}

  阅读代码后，用户头脑中 job 的执行流程是这样的： 具体流程很简单，这里来估算下 data size 和执行结果：

1. 初始化 SparkConf()。
2. 初始化 numMappers=100, numKVPairs=10,000, valSize=1000, numReducers= 36。
3. 初始化 SparkContext。这一步很重要，是要确立 driver 的地位，里面包含建立 driver 所需的各类 actors 和 objects。
4. 每一个 mapper 生成一个 arr1: Array[(Int, Byte[])]，length 为 numKVPairs。每个 Byte[] 的 length 为 valSize，Int 为随机生成的整数。Size(arr1) = numKVPairs * (4 + valSize) = 10MB，因此Size(pairs1) = numMappers * Size(arr1) ＝1000MB。这里的数值计算结果都是约等于。
5. 每一个 mapper 将产生的 arr1 数组 cache 到内存。
6. 而后执行一个 action 操做 count()，来统计全部 mapper 中 arr1 中的元素个数，执行结果是numMappers * numKVPairs = 1,000,000。这一步主要是为了将每一个 mapper 产生的 arr1 数组 cache 到内存。
7. 在已经被 cache 的 paris1 上执行 groupByKey 操做，groupByKey 产生的 reducer （也就是 partition） 个数为 numReducers。理论上，若是 hash(Key) 比较平均的话，每一个 reducer 收到的<int, array[byte]="">record 个数为 numMappers * numKVPairs / numReducer ＝ 27,777，大小为 Size(pairs1) / numReducer = 27MB。
8. reducer 将收到的 <Int, Byte[]> records 中拥有相同 Int 的 records 聚在一块儿，获得 <Int, list(Byte[], Byte[], ..., Byte[])>。
9. 最后 count 将全部 reducer 中 records 个数进行加和，最后结果实际就是 pairs1 中不一样的 Int 总个数。

Job 逻辑执行图

Job 的实际执行流程比用户头脑中的要复杂，须要先创建逻辑执行图（或者叫数据依赖图），而后划分逻辑执行图生成 DAG 型的物理执行图，而后生成具体 task 执行。分析一下这个 job 的逻辑执行图： 使用 RDD.toDebugString 能够看到整个 logical plan （RDD 的数据依赖关系）以下

MapPartitionsRDD[3] at groupByKey at GroupByTest.scala:51 (36 partitions)
    ShuffledRDD[2] at groupByKey at GroupByTest.scala:51 (36 partitions)
      FlatMappedRDD[1] at flatMap at GroupByTest.scala:38 (100 partitions)
        ParallelCollectionRDD[0] at parallelize at GroupByTest.scala:38 (100 partitions)

  用图表示就是：

须要注意的是 data in the partition 展现的是每一个 partition 应该获得的计算结果，并不意味着这些结果都同时存在于内存中。

根据上面的分析可知：

• 用户首先 init 了一个0-99 的数组： 0 until numMappers
• parallelize() 产生最初的 ParrallelCollectionRDD，每一个 partition 包含一个整数 i。
• 执行 RDD 上的 transformation 操做（这里是 flatMap）之后，生成 FlatMappedRDD，其中每一个 partition 包含一个 Array[(Int, Array[Byte])]。
• 第一个 count() 执行时，先在每一个 partition 上执行 count，而后执行结果被发送到 driver，最后在 driver 端进行 sum。
• 因为 FlatMappedRDD 被 cache 到内存，所以这里将里面的 partition 都换了一种颜色表示。
• groupByKey 产生了后面两个 RDD，为何产生这两个在后面章节讨论。
• 若是 job 须要 shuffle，通常会产生 ShuffledRDD。该 RDD 与前面的 RDD 的关系相似于 Hadoop 中 mapper 输出数据与 reducer 输入数据之间的关系。
• MapPartitionsRDD 里包含 groupByKey() 的结果。
• 最后将 MapPartitionsRDD 中的 每一个value（也就是Array[Byte]）都转换成 Iterable 类型。
• 最后的 count 与上一个 count 的执行方式相似。

能够看到逻辑执行图描述的是 job 的数据流：job 会通过哪些 transformation()，中间生成哪些 RDD 及 RDD 之间的依赖关系。

Job 物理执行图

逻辑执行图表示的是数据上的依赖关系，不是 task 的执行图。在 Hadoop 中，用户直接面对 task，mapper 和 reducer 的职责分明：一个进行分块处理，一个进行 aggregate。Hadoop 中 整个数据流是固定的，只须要填充 map() 和 reduce() 函数便可。Spark 面对的是更复杂的数据处理流程，数据依赖更加灵活，很难将数据流和物理 task 简单地统一在一块儿。所以 Spark 将数据流和具体 task 的执行流程分开，并设计算法将逻辑执行图转换成 task 物理执行图，转换算法后面的章节讨论。 针对这个 job，咱们先画出它的物理执行 DAG 图以下： 能够看到 GroupByTest 这个 application 产生了两个 job，第一个 job 由第一个 action（也就是 pairs1.count）触发产生，分析一下第一个 job：

• 整个 job 只包含 1 个 stage（不明白什么是stage不要紧，后面章节会解释，这里只需知道有这样一个概念）。
• Stage 0 包含 100 个 ResultTask。
• 每一个 task 先计算 flatMap，产生 FlatMappedRDD，而后执行 action() 也就是 count()，统计每一个 partition 里 records 的个数，好比 partition 99 里面只含有 9 个 records。
• 因为 pairs1 被声明要进行 cache，所以在 task 计算获得 FlatMappedRDD 后会将其包含的 partitions 都 cache 到 executor 的内存。
• task 执行完后，driver 收集每一个 task 的执行结果，而后进行 sum()。
• job 0 结束。

第二个 job 由 pairs1.groupByKey(numReducers).count 触发产生。分析一下该 job：

• 整个 job 包含 2 个 stage。
• Stage 1 包含 100 个 ShuffleMapTask，每一个 task 负责从 cache 中读取 pairs1 的一部分数据并将其进行相似 Hadoop 中 mapper 所作的 partition，最后将 partition 结果写入本地磁盘。
• Stage 0 包含 36 个 ResultTask，每一个 task 首先 shuffle 本身要处理的数据，边 fetch 数据边进行 aggregate 以及后续的 mapPartitions() 操做，最后进行 count() 计算获得 result。
• task 执行完后，driver 收集每一个 task 的执行结果，而后进行 sum()。
• job 1 结束。

能够看到物理执行图并不简单。与 MapReduce 不一样的是，Spark 中一个 application 可能包含多个 job，每一个 job 包含多个 stage，每一个 stage 包含多个 task。 怎么划分 job，怎么划分 stage，怎么划分 task 等等问题会在后面的章节介绍。

Discussion

到这里，咱们对整个系统和 job 的生成与执行有了概念，并且还探讨了 cache 等特性。 接下来的章节会讨论 job 生成与执行涉及到的系统核心功能，包括：

1. 如何生成逻辑执行图
2. 如何生成物理执行图
3. 如何提交与调度 Job
4. Task 如何生成、执行与结果处理
5. 如何进行 shuffle
6. cache机制
7. broadcast 机制

本文转载于：http://spark-internals.books.yourtion.com/