Spark机制原理
本书前面几章分别介绍了Spark的生态系统、Spark运行模式及Spark的核心概念RDD和基本算子操作等重要基础知识。本章重点讲解Spark的主要机制原理，因为这是Spark程序得以高效执行的核心。本章先从Application、job、stage和task等层次阐述Spark的调度逻辑，并且介绍FIFO、FAIR等经典算法，然后对Spark的重要组成模块：I/O与通信控制模块、容错模块及Shuffle模块做了深入的阐述。其中，在Spark I/O模块中，数据以数据块的形式管理，存储在内存、磁盘或者Spark集群中的其他机器上。Spark集群通信机制采用了AKKA通信框架，在集群机器中传递命令和状态信息。另外，容错是分布式系统的一个重要特性，Spark采用了lineage与checkpoint机制来保证容错性。Spark Shuffle模块借鉴了MapReduce的Shuffle机制，但在其基础上进行了改进与创新。
3.1 Spark应用执行机制分析
下面对Spark Application的基本概念和执行机制进行深入介绍。
3.1.1 Spark应用的基本概念
Spark应用（Application）是用户提交的应用程序。Spark运行模式分为：Local、Standalone、YARN、Mesos等。根据Spark Application的Driver Program是否在集群中运行，Spark应用的运行方式又可以分为Cluster模式和Client模式。
下面介绍Spark应用涉及的一些基本概念：
1） SparkContext：Spark 应用程序的入口，负责调度各个运算资源，协调各个Worker Node 上的Executor。
2） Driver Program：运行Application的main()函数并创建SparkContext。
3） RDD：前面已经讲过，RDD是Spark的核心数据结构，可以通过一系列算子进行操作。当RDD遇到Action算子时，将之前的所有算子形成一个有向无环图（DAG）。再在Spark中转化为Job（Job的概念在后面讲述），提交到集群执行。一个App中可以包含多个Job。
4） Worker Node：集群中任何可以运行Application代码的节点，运行一个或多个Executor进程。
5） Executor：为Application运行在Worker Node上的一个进程，该进程负责运行Task，并且负责将数据存在内存或者磁盘上。每个Application都会申请各自的Executor来处理任务。
下面介绍Spark 应用（Application）执行过程中各个组件的概念：
1） Task（任务）：RDD中的一个分区对应一个Task，Task是单个分区上最小的处理流程单元。
2） TaskSet（任务集）： 一组关联的，但相互之间没有Shuffle依赖关系的Task集合。
3） Stage（调度阶段）：一个TaskSet对应的调度阶段。每个Job会根据RDD的宽依赖关系被切分很多Stage，每个Stage都包含一个TaskSet。
4） Job（作业）： 由Action算子触发生成的由一个或多个Stage组成的计算作业。
5） Application：用户编写的Spark的应用程序，由一个或多个Job组成。提交到Spark之后，Spark为Application分配资源，将程序转换并执行。
6） DAGScheduler：根据Job构建基于Stage的DAG，并提交Stage给TaskScheduler。
7） TaskScheduler：将Taskset提交给Worker Node集群运行并返回结果。
以上基本概念之间的关系如图3-1所示。
3.1.2 Spark应用执行机制概要
Spark Application从提交后到在Worker Node执行，期间经历了一系列变换，具体过程如图3-2所示。
图3-1 Spark基本概念之间的关系
图3-2 Spark 执行流程
如图3-2所示，前面讲过，当RDD遇见Action算子之后，触发Job提交。提交后的Job在Spark中形成了RDD DAG有向无环图（Directed Acyclic Graph）。RDD DAG经过DAG Scheduler调度之后，根据RDD依赖关系被切分为一系列的Stage。每个Stage包含一组task集合，再经过Task Scheduler之后，task被分配到Worker节点上的Executor线程池执行。如前文所述，RDD中的每一个逻辑分区对应一个物理的数据块，同时每个分区对应一个Task，因此Task也有自己对应的物理数据块，使用用户定义的函数来处理。Spark出于节约内存的考虑，采用了延迟执行的策略，如前文所述，只有Action算子才可以触发整个操作序列的执行。另外，Spark对于中间计算结果也不会重新分配内存，而是在同一个数据块上流水线操作。
Spark使用BlockManager管理数据块，在内存或者磁盘进行存储，如果数据不在本节点，则还可以通过远端节点复制到本机进行计算。在计算时，Spark会在具体执行计算的Worker节点的Executor中创建线程池，Executor将需要执行的任务通过线程池来并发执行。
3.1.3 应用提交与执行
Spark使用Driver进程负责应用的解析、切分Stage并调度Task到Executor执行，包含DAGScheduler等重要对象。Driver进程的运行地点有如下两种：
1） Driver进程运行在Client端，对应用进行管理监控。
2） Master节点指定某个Worker节点启动Driver进程，负责监控整个应用的执行。
针对这两种情况，应用提交及执行过程分别如下：
1. Driver运行在Client
用户启动Client端，在Client端启动Driver进程。在Driver中启动或实例化DAGS-
cheduler等组件。
1）Driver向Master注册。
2）Worker向Master注册，Master通过指令让Worker启动Executor。
3）Worker通过创建ExecutorRunner线程，进而ExecutorRunner线程启动Executor-Backend进程。
4）ExecutorBackend启动后，向Client端Driver进程内的SchedulerBackend注册，因此Driver进程就可以发现计算资源。
5）Driver的DAGScheduler解析应用中的RDD DAG并生成相应的Stage，每个Stage包含的TaskSet通过TaskScheduler分配给Executor。在Executor内部启动线程池并行化执行Task。
2. Driver运行在Worker节点
用户启动客户端，客户端提交应用程序给Master。
1）Master调度应用，指定一个Worker节点启动Driver，即Scheduler-Backend。
2）Worker接收到Master命令后创建DriverRunner线程，在DriverRunner线程内创建SchedulerBackend进程。Driver充当整个作业的主控进程。
3）Master指定其他Worker节点启动Exeuctor，此处流程和上面相似，Worker创建ExecutorRunner线程，启动ExecutorBackend进程。
4）ExecutorBackend启动后，向Driver的SchedulerBackend注册，这样Driver获取了计算资源就可以调度和将任务分发到计算节点执行。
SchedulerBackend进程中包含DAGScheduler，它会根据RDD的DAG切分Stage，生成TaskSet，并调度和分发Task到Executor。对于每个Stage的TaskSet，都会被存放到TaskScheduler中。TaskScheduler将任务分发到Executor，执行多线程并行任务。
图3-3为Spark应用的提交与执行示意图。
图3-3 Spark应用的提交与执行
3.2 Spark调度机制
Spark调度机制是保证Spark应用高效执行的关键。本节从Application、job、stage和task的维度，从上层到底层来一步一步揭示Spark的调度策略。
3.2.1 Application的调度
Spark中，每个Application对应一个SparkContext。SparkContext之间的调度关系取决于Spark的运行模式。对Standalone模式而言，Spark Master节点先计算集群内的计算资源能否满足等待队列中的应用对内存和CPU资源的需求，如果可以，则Master创建Spark Driver，启动应用的执行。宏观上来讲，这种对应用的调度类似于FIFO策略。在Mesos和YARN模式下，底层的资源调度系统的调度策略都是由Mesos和YARN决定的。具体分类描述如下：
1. Standalone模式
默认以用户提交Application的顺序来调度，即FIFO策略。每个应用执行时独占所有资源。如果有多个用户要共享集群资源，则可以使用参数spark.cores.max来配置应用在集群中可以使用的最大CPU核数。如果不配置，则采用默认参数spark.deploy.defaultCore的值来确定。
2. Mesos模式
如果在Mesos上运行Spark，用户想要静态配置资源的话，可以设置spark.mesos.coarse为true，这样Mesos变为粗粒度调度模式，然后可以设置spark.cores.max指定集群中可以使用的最大核数，与上面的Standalone模式类似。同时，在Mesos模式下，用户还可以设置参数spark.executor.memory来配置每个executor的内存使用量。如果想使Mesos在细粒度模式下运行，可以通过mesos://<url-info>设置动态共享cpu core的执行模式。在这种模式下，应用不执行时的空闲CPU资源得以被其他用户使用，提升了CPU使用率。
3. YARN模式
如果在YARN上运行Spark，用户可以在YARN的客户端上设置--num-executors 来控制为应用分配的Executor数量，然后设置--executor-memory指定每个Executor的内存大小，设置--executor-cores指定Executor占用的CPU核数。
3.2.2 job的调度
前面章节提到过，Spark应用程序实际上是一系列对RDD的操作，这些操作直至遇见Action算子，才触发Job的提交。事实上，在底层实现中，Action算子最后调用了runJob函数提交Job给Spark。其他的操作只是生成对应的RDD关系链。如在RDD.scala程序文件中，count函数源码所示。
def count(): Long = sc.runJob(this, Utils.getIteratorSize _).sum
其中sc为SparkContext的对象。可见在Spark中，对Job的提交都是在Action算子中隐式完成的，并不需要用户显式地提交作业。在SparkContext中Job提交的实现中，最后会调用DAGScheduler中的Job提交接口。DAGScheduler最重要的任务之一就是计算Job与Task的依赖关系，制定调度逻辑。
Job调度的基本工作流程如图3-4所示，每个Job从提交到完成，都要经历一系列步骤，拆分成以Tsk为最小单位，按照一定逻辑依赖关系的执行序列。
图3-4 Job的调度流程
图3-5则从Job调度流程中的细节模块出发，揭示了工作流程与对应模块之间的关系。从整体上描述了各个类在Job调度流程中的交互关系。
图3-5 Job调度流程细节
在Spark1.5.0的调度目录下的SchedulingAlgorithm.scala文件中，描述了Spark对Job的调度模式。
1. FIFO模式
默认情况下，Spark对Job以FIFO（先进先出）的模式进行调度。在SchedulingAlgorithm.scala文件中声明了FIFO算法实现。
private[spark] class FIFOSchedulingAlgorithm extends SchedulingAlgorithm {
override def comparator(s1: Schedulable, s2: Schedulable): Boolean = {
//定义优先级
val priority1 = s1.priority
val priority2 = s2.priority
var res = math.signum(priority1 - priority2)
if (res == 0) {
val stageId1 = s1.stageId
val stageId2 = s2.stageId
//signum是符号函数,返回0(参数等于0)、1(参数大于0)或-1(参数小于0)。
res = math.signum(stageId1 - stageId2)
}
if (res < 0) {
true
} else {
false
}
}
}
2. FAIR模式
Spark在FAIR的模式下，采用轮询的方式为多个Job分配资源，调度Job。所有的任务优先级大致相同，共享集群计算资源。具体实现代码在SchedulingAlgorithm.scala文件中，声明如下：
private[spark] class FairSchedulingAlgorithm extends SchedulingAlgorithm {
override def comparator(s1: Schedulable, s2: Schedulable): Boolean = {
val minShare1 = s1.minShare
val minShare2 = s2.minShare
val runningTasks1 = s1.runningTasks
val runningTasks2 = s2.runningTasks
val s1Needy = runningTasks1 < minShare1
val s2Needy = runningTasks2 < minShare2
val minShareRatio1 = runningTasks1.toDouble / math.max(minShare1, 1.0).toDouble
val minShareRatio2 = runningTasks2.toDouble / math.max(minShare2, 1.0).toDouble
val taskToWeightRatio1 = runningTasks1.toDouble / s1.weight.toDouble
val taskToWeightRatio2 = runningTasks2.toDouble / s2.weight.toDouble
var compare: Int = 0
if (s1Needy && !s2Needy) {
return true
} else if (!s1Needy && s2Needy) {
return false
} else if (s1Needy && s2Needy) {
compare = minShareRatio1.compareTo(minShareRatio2)
} else {
compare = taskToWeightRatio1.compareTo(taskToWeightRatio2)
}
if (compare < 0) {
true
} else if (compare > 0) {
false
} else {
s1.name < s2.name
}
}
}
3. 配置调度池
DAGScheduler构建了具有依赖关系的任务集。TaskScheduler负责提供任务给Task-SetManager作为调度的先决条件。TaskSetManager负责具体任务集内部的调度任务。调度池（pool）则用于调度每个SparkContext运行时并存的多个互相独立无依赖关系的任务集。调度池负责管理下一级的调度池和TaskSetManager对象。
用户可以通过配置文件定义调度池的属性。一般调度池支持如下3个参数：
1）调度模式Scheduling mode：用户可以设置FIFO或者FAIR调度方式。
2）weight：调度池的权重，在获取集群资源上权重高的可以获取多个资源。
3）miniShare：代表计算资源中的CPU核数。
用户可以通过conf/fairscheduler.xml配置调度池的属性，同时要在SparkConf对象中配置属性。
3.2.3 stage（调度阶段）和TasksetManager的调度
1. Stage划分
当一个Job被提交后，DAGScheduler会从RDD依赖链的末端触发，遍历整个RDD依赖链，划分Stage（调度阶段）。划分依据主要基于ShuffleDependency依赖关系。换句话说，当某RDD在计算中需要将数据进行Shuffle操作时，这个包含Shuffle操作的RDD将会被用来作为输入信息，构成一个新的Stage。以这个基准作为划分Stage，可以保证存在依赖关系的数据按照正确数据得到处理和运算。在Spark1.5.0的源代码中，DAGScheduler.scala中的getParentStages函数的实现从一定角度揭示了Stage的划分逻辑。
/**
* 对于给定的RDD构建或获取父Stage的链表。新的Stage构建时会包含参数中提供的firstJobId
*/
private def getParentStages(rdd: RDD[_], firstJobId: Int): List[Stage] = {
val parents = new HashSet[Stage]
val visited = new HashSet[RDD[_]]
// We are manually maintaining a stack here to prevent StackOverflowError
// caused by recursively visiting
val waitingForVisit = new Stack[RDD[_]]
def visit(r: RDD[_]) {
if (!visited(r)) {
visited += r
// Kind of ugly: need to register RDDs with the cache here since
// we can't do it in its constructor because # of partitions is unknown
/* 遍历RDD的依赖链 */
for (dep <- r.dependencies) {
dep match {
/*如果遇见ShuffleDependency，则依据此依赖关系划分Stage，并添加该Stage的父Stage到哈希列表中*/
case shufDep: ShuffleDependency[_, _, _] =>
parents += getShuffleMapStage(shufDep, firstJobId)
case _ =>
waitingForVisit.push(dep.rdd)
}
}
}
}
2. Stage调度
在第一步的Stage划分过程中，会产生一个或者多个互相关联的Stage。其中，真正执行Action算子的RDD所在的Stage被称为Final Stage。DAGScheduler会从这个final stage生成作业实例。
在Stage提交时，DAGScheduler首先会判断该Stage的父Stage的执行结果是否可用。如果所有父Stage的执行结果都可用，则提交该Stage。如果有任意一个父Stage的结果不可用，则尝试迭代提交该父Stage。所有结果不可用的Stage都将会被加入waiting队列，等待执行，如图3-6所示。
图3-6 Stage依赖
在图3-6中，虚箭头表示依赖关系。Stage序号越小，表示Stage越靠近上游。
图3-6中的Stage调度运行顺序如图3-7所示。
图3-7 Stage执行顺序
从图3-7可以看出，上游父Stage先得到执行，waiting queue中的stage随后得到执行。
3. TasksetManager
每个Stage的提交会被转化为一组task的提交。DAGScheduler最终通过调用taskscheduler的接口来提交这组任务。在taskScheduler内部实现中创建了taskSetManager实例来管理任务集taskSet的生命周期。事实上可以说每个stage对应一个tasksetmanager。
至此，DAGScheduler的工作基本完毕。taskScheduler在得到集群计算资源时，taskSet-Manager会分配task到具体worker节点上执行。在Spark1.5.0的taskSchedulerImpl.scala文件中，提交task的函数实现如下：
override def submitTasks(taskSet: TaskSet) {
val tasks = taskSet.tasks
logInfo("Adding task set " + taskSet.id + " with " + tasks.length + " tasks")
this.synchronized {
/*创建TaskSetManager实例以管理stage包含的任务集*/
val manager = createTaskSetManager(taskSet, maxTaskFailures)
val stage = taskSet.stageId
val stageTaskSets =
taskSetsByStageIdAndAttempt.getOrElseUpdate(stage, new HashMap[Int, TaskSetManager])
stageTaskSets(taskSet.stageAttemptId) = manager
val conflictingTaskSet = stageTaskSets.exists { case (_, ts) =>
ts.taskSet != taskSet && !ts.isZombie
}
if (conflictingTaskSet) {
throw new IllegalStateException(s"more than one active taskSet for stage $stage:" +
s" ${stageTaskSets.toSeq.map{_._2.taskSet.id}.mkString(",")}")
}
/*将TaskSetManager添加到全局的调度队列*/
schedulableBuilder.addTaskSetManager(manager, manager.taskSet.properties)
if (!isLocal && !hasReceivedTask) {
starvationTimer.scheduleAtFixedRate(new TimerTask() {
override def run() {
if (!hasLaunchedTask) {
logWarning("Initial job has not accepted any resources; " +
"check your cluster UI to ensure that workers are registered " +
"and have sufficient resources")
} else {
this.cancel()
}
}
}, STARVATION_TIMEOUT_MS, STARVATION_TIMEOUT_MS)
}
hasReceivedTask = true
}
backend.reviveOffers()
}
当taskSetManager进入到调度池中时，会依据job id对taskSetManager排序，总体上先进入的taskSetManager先得到调度。对于同一job内的taskSetManager而言，job id较小的先得到调度。如果有的taskSetManager父Stage还未执行完，则该taskSet-Manager不会被放到调度池。
3.2.4 task的调度
在DAGScheduler.scala中，定义了函数submitMissingTasks，读者阅读完整实现，从中可以看到task的调度方式。限于篇幅，以下截取部分代码。
private def submitMissingTasks(stage: Stage, jobId: Int) {
logDebug("submitMissingTasks(" + stage + ")")
// Get our pending tasks and remember them in our pendingTasks entry
stage.pendingTasks.clear()
// First figure out the indexes of partition ids to compute.
/*过滤出计算位置，用以执行计算*/
val (allPartitions: Seq[Int], partitionsToCompute: Seq[Int]) = {
stage match {
/*针对shuffleMap类型的Stage*/
case stage: ShuffleMapStage =>
val allPartitions = 0 until stage.numPartitions
val filteredPartitions = allPartitions.filter { id => stage.outputLocs(id).isEmpty }
(allPartitions, filteredPartitions)
/*针对Result类型的Stage*/
case stage: ResultStage =>
val job = stage.resultOfJob.get
val allPartitions = 0 until job.numPartitions
val filteredPartitions = allPartitions.filter { id => ! job.finished(id) }
(allPartitions, filteredPartitions)
}
}
.....[以下代码略]
/*获取task执行的优先节点*/
private[spark]
def getPreferredLocs(rdd: RDD[_], partition: Int): Seq[TaskLocation] = {
getPreferredLocsInternal(rdd, partition, new HashSet)
}
计算task执行的优先节点位置的代码实现在getPreferredLocsInternal函数中，具体如下：
/*计算位置的递归实现*/
private def getPreferredLocsInternal(
rdd: RDD[_],
partition: Int,
visited: HashSet[(RDD[_], Int)]): Seq[TaskLocation] = {
// If the partition has already been visited, no need to re-visit.
// This avoids exponential path exploration. SPARK-695
if (!visited.add((rdd, partition))) {
// Nil has already been returned for previously visited partitions.
return Nil
}
// 如果调用cache缓存过，则计算缓存位置，读取缓存分区中的数据
val cached = getCacheLocs(rdd)(partition)
if (cached.nonEmpty) {
return cached
}
// 如果能直接获取到执行地点，则返回作为该task的执行地点
val rddPrefs = rdd.preferredLocations(rdd.partitions(partition)).toList
if (rddPrefs.nonEmpty) {
return rddPrefs.map(TaskLocation(_))
}
/*针对窄依赖关系的RDD, 取出第一个窄依赖的父RDD分区的执行地点*/
rdd.dependencies.foreach {
case n: NarrowDependency[_] =>
for (inPart <- n.getParents(partition)) {
val locs = getPreferredLocsInternal(n.rdd, inPart, visited)
if (locs != Nil) {
return locs
}
}
case _ =>
}
/*对于shuffle依赖的rdd，选取至少含REDUCER_PREF_LOCS_FRACTION这么多数据的位置作为优先节点*/
if (shuffleLocalityEnabled && rdd.partitions.length < SHUFFLE_PREF_REDUCE_THRESHOLD) {
rdd.dependencies.foreach {
case s: ShuffleDependency[_, _, _] =>
if (s.rdd.partitions.length < SHUFFLE_PREF_MAP_THRESHOLD) {
// Get the preferred map output locations for this reducer
val topLocsForReducer = mapOutputTracker.getLocationsWithLargestOu-tputs(s.shuffleId,
partition, rdd.partitions.length, REDUCER_PREF_LOCS_FRACTION)
if (topLocsForReducer.nonEmpty) {
return topLocsForReducer.get.map(loc => TaskLocation(loc.host, loc.executorId))
}
}
case _ =>
}
}
Nil
}
3.3 Spark存储与I/O
前面已经讲过，RDD是按照partition分区划分的，所以RDD可以看作由一些分布在不同节点上的分区组成。由于partition分区与数据块是一一对应的，所以RDD中保存了partitionID与物理数据块之间的映射。物理数据块并非都保存在磁盘上，也有可能保存在内存中。
3.3.1 Spark存储系统概览
Spark I/O机制可以分为两个层次：
1）通信层：用于Master与Slave之间传递控制指令、状态等信息，通信层在架构上也采用Master-Slave结构。
2）存储层：同于保存数据块到内存、磁盘，或远端复制数据块。
下面介绍几个Spark存储方面的功能模块。
1）BlockManager：Spark提供操作Storage的统一接口类。
2）BlockManagerMasterActor：Master创建，Slave利用该模块向Master传递信息。
3）BlockManagerSlaveActor：Slave创建，Master利用该模块向Slave节点传递控制命令，控制Slave节点对block的读写。
4）BlockManagerMaster： 管理Actor通信。
5）DiskStore：支持以文件方式读写的方式操作block。
6）MemoryStore： 支持内存中的block读写。
7）BlockManagerWorker: 对远端异步传输进行管理。
8）ConnectionManager：支持本地节点与远端节点数据block的传输。
图3-8概要性地揭示了Spark存储系统各个主要模块之间的通信。
图3-8 Spark存储系统概览
3.3.2 BlockManager中的通信
存储系统的通信仍然类似Master-Slave架构，节点之间传递命令与状态。总体而言，Master向Slave传递命令，Slave向Master传递信息和状态。这些Master与Slave节点之间的信息传递通过Actor对象实现（关于Actor的详细功能会在下一节Spark通信机制中讲述）。但在BlockManager中略有不同，下面分别讲述。
1）Master节点上的BlockManagerMaster包含内容如下：
①BlockManagerMasterActor的Actor引用。
②BlockManagerSlaveActor的Ref引用。
2）Slave节点上的BlockManagerMaster包含内容如下：
①BlockManagerMasterActor的Ref引用。
②BlockManagerSlaveActor的Actor引用。
其中，在Ref与Actor之间的通信由BlockManagerMasterActor和BlockManagerSlave-Actor完成。这个部分相关的源码篇幅较多，此处省略，感兴趣的读者请自行研究。
3.4 Spark通信机制
前面介绍过，Spark的部署模式可以分为local、standalone、Mesos、YARN等。
本节以Spark部署在standalone模式下为例，介绍Spark的通信机制（其他模式类似）。
3.4.1 分布式通信方式
先介绍分布式通信的几种基本方式。
1. RPC
远程过程调用协议（Remote Procedure Call Protocol，RPC）是一种通过网络从远程计算机程序上请求服务，而不需要了解底层网络技术的协议。RPC假定某些传输协议的存在，如TCP或UDP，为通信程序之间携带信息数据。在OSI网络通信模型中，RPC跨越了传输层和应用层。RPC使得开发分布式应用更加容易。RPC采用C/S架构。请求程序就是一个Client，而服务提供程序就是一个Server。首先，Client调用进程发送一个有进程参数的调用信息到Service进程，然后等待应答信息。在Server端，进程保持睡眠状态直到调用信息到达为止。当一个调用信息到达时，Server获得进程参数，计算结果，发送答复信息，然后等待下一个调用信息，最后，Client调用进程接收答复信息，获得进程结果，然后调用执行继续进行。
2. RMI
远程方法调用（Remote Method Invocation，RMI）是Java的一组拥护开发分布式应用程序的API。RMI使用Java语言接口定义了远程对象，它集合了Java序列化和Java远程方法协议（Java Remote Method Protocol）。简单地说，这样使原先的程序在同一操作系统的方法调用，变成了不同操作系统之间程序的方法调用。由于J2EE是分布式程序平台，它以RMI机制实现程序组件在不同操作系统之间的通信。比如，一个EJB可以通过RMI调用Web上另一台机器上的EJB远程方法。RMI可以被看作是RPC的Java版本，但是传统RPC并不能很好地应用于分布式对象系统。Java RMI 则支持存储于不同地址空间的程序级对象之间彼此进行通信，实现远程对象之间的无缝远程调用。
3. JMS
Java消息服务（Java Message Service，JMS）是一个与具体平台无关的API，用来访问消息收发。JMS 使用户能够通过消息收发服务（有时称为消息中介程序或路由器）从一个 JMS 客户机向另一个JMS客户机发送消息。消息是 JMS 中的一种类型对象，由两部分组成：报头和消息主体。报头由路由信息以及有关该消息的元数据组成。消息主体则携带着应用程序的数据或有效负载。JMS定义了5种消息正文格式，以及调用的消息类型，允许发送并接收以一些不同形式的数据，提供现有消息格式的一些级别的兼容性。
StreamMessage：Java原始值的数据流。
MapMessage：一套名称–值对。
TextMessage：一个字符串对象。
ObjectMessage：一个序列化的 Java对象。
BytesMessage：一个未解释字节的数据流。
4. EJB
JavaEE服务器端组件模型（Enterprise JavaBean，EJB）的设计目标是部署分布式应用程序。简单来说就是把已经编写好的程序打包放在服务器上执行。EJB定义了一个用于开发基于组件的企业多重应用程序的标准。EJB的核心是会话Bean（Session Bean）、实体Bean（Entity Bean）和消息驱动Bean（Message Driven Bean）。
5. Web Service
Web Service是一个平台独立的、低耦合的、自包含的、基于可编程的Web应用程序。可以使用开放的XML（标准通用标记语言下的一个子集）标准来描述、发布、发现、协调和配置这些应用程序，用于开发分布式的应用程序。Web Service技术能使得运行在不同机器上的不同应用无须借助第三方软硬件， 就可相互交换数据或集成。Web Service减少了应用接口的花费。Web Service为整个企业甚至多个组织之间的业务流程的集成提供了一个通用机制。
3.4.2 通信框架AKKA
AKKA是一个用Scala语言编写的库，用于简化编写容错的、高可伸缩性的Java和Scala的Actor模型应用。它分为开发库和运行环境，可以用于构建高并发、分布式、可容错、事件驱动的基于JVM的应用。AKKA使构建高并发的分布式应用变得更加容易。Akka已经被成功运用在众多行业的众多大企业，从投资业到商业银行、从零售业到社会媒体、仿真、游戏和赌博、汽车和交通系统、数据分析等。任何需要高吞吐率和低延迟的系统都是使用AKKA的候选，因此Spark选择AKKA通信框架来支持模块间的通信。
Actor模型常见于并发编程，它由Carl Hewitt于20世纪70年代早期提出，目的是解决分布式编程中的一系列问题。其特点如下：
1） 系统中的所有事物都可以扮演一个Actor。
2） Actor之间完全独立。
3） 在收到消息时Actor采取的所有动作都是并行的。
4） Actor有标识和对当前行为的描述。
Actor可以看作是一个个独立的实体，它们之间是毫无关联的。但是，它们可以通过消息来通信。当一个Actor收到其他Actor的信息后，它可以根据需要做出各种响应。消息的类型和内容都可以是任意的。这点与Web Service类似，只提供接口服务，不必了解内部实现。一个Actor在处理多个Actor的请求时，通常先建立一个消息队列，每次收到消息后，就放入队列。Actor每次也可以从队列中取出消息体来处理，而且这个过程是可循环的，这个特点让Actor可以时刻处理发送来的消息。
AKKA的优势如下：
1） 易于构建并行与分布式应用（simple concurrency & distribution）：AKKA采用异步通信与分布式架构，并对上层进行抽象，如Actors、Futures、STM等。
2） 可靠性（resilient by design）：系统具备自愈能力，在本地/远程都有监护。
3） 高性能（high performance）：在单机中每秒可发送5000万个消息。内存占用小，1GB内存中可保存250万个actors。
4） 弹性，无中心（elastic — decentralized）：自适应的负责均衡、路由、分区、配置。
5） 可扩展性（extensible）：可以使用Akka扩展包进行扩展。
3.4.3 Client、Master 和 Worker之间的通信
Client、Master与Worker之间的交互代码实现位于如下路径：
(spark-root)/core/src/main/scala/org/apache/spark/deploy
主要涉及的类包括Client.scala、Master.scala和Worker.scala。这三大模块之间的通信框架如图3-9所示：
图3-9 Client、Master和Worker之间的通信
以Standalone部署模式为例，三大模块分工如下：
1）Client：提交作业给Master。
2）Master：接收Client提交的作业，管理Worker，并命令Worker启动Driver和Executor。
3）Worker：负责管理本节点的资源，定期向Master汇报心跳信息，接收Master的命令，如启动Driver和Executor。
下面列出Client、Master与Worker的实现代码，读者可以从中看到三个模块间的通信交互。
1. Client端通信
private class ClientEndpoint(
override val rpcEnv: RpcEnv,
driverArgs: ClientArguments,
masterEndpoints: Seq[RpcEndpointRef],
conf: SparkConf)
extends ThreadSafeRpcEndpoint with Logging {
<限于篇幅，此处代码省略……>
override def onStart(): Unit = {
driverArgs.cmd match {
case "launch" =>
val mainClass = "org.apache.spark.deploy.worker.DriverWrapper"
val classPathConf = "spark.driver.extraClassPath"
val classPathEntries = sys.props.get(classPathConf).toSeq.flatMap { cp =>
cp.split(java.io.File.pathSeparator)
}
val libraryPathConf = "spark.driver.extraLibraryPath"
val libraryPathEntries = sys.props.get (libraryPathConf).toSeq.flatMap { cp =>
cp.split(java.io.File.pathSeparator)
}
val extraJavaOptsConf = "spark.driver.extraJavaOptions"
val extraJavaOpts = sys.props.get(extraJavaOptsConf)
.map(Utils.splitCommandString).getOrElse(Seq.empty)
val sparkJavaOpts = Utils.sparkJavaOpts(conf)
val javaOpts = sparkJavaOpts ++ extraJavaOpts
val command = new Command(mainClass,
Seq("{{WORKER_URL}}", "{{USER_JAR}}", driverArgs.mainClass) ++ driverArgs.driverOptions,
sys.env, classPathEntries, libraryPathEntries, javaOpts)
/* 创建driverDescription对象 */
val driverDescription = new DriverDescription(
driverArgs.jarUrl,
driverArgs.memory,
driverArgs.cores,
driverArgs.supervise,
command)
/* 此处向Master的Actor提交Driver*/
ayncSendToMasterAndForwardReply[SubmitDriverResponse](
RequestSubmitDriver(driverDescription))
case "kill" =>
val driverId = driverArgs.driverId
/* 接收停止Driver是否成功的通知 */
ayncSendToMasterAndForwardReply[KillDriverResponse](RequestKill-Driver(driverId))
}
}
/* 向Master发送消息，并异步地转发返回信息给Client */
private def ayncSendToMasterAndForwardReply[T: ClassTag](message: Any): Unit = {
for (masterEndpoint <- masterEndpoints) {
masterEndpoint.ask[T](message).onComplete {
case Success(v) => self.send(v)
case Failure(e) =>
logWarning(s"Error sending messages to master $masterEndpoint", e)
}(forwardMessageExecutionContext)
}
}
2. Master端通信
private[deploy] class Master(
override val rpcEnv: RpcEnv,
address: RpcAddress,
webUiPort: Int,
val securityMgr: SecurityManager,
val conf: SparkConf)
extends ThreadSafeRpcEndpoint with Logging with LeaderElectable {
……
override def receive: PartialFunction[Any, Unit] = {
/* 选举为Master，当状态为RecoveryState.RECOVERING时恢复 */
case ElectedLeader => {
val (storedApps, storedDrivers, storedWorkers) = persistenceEngine.readPersistedData(rpcEnv)
state = if (storedApps.isEmpty && storedDrivers.isEmpty && storedWorkers.isEmpty) {
RecoveryState.ALIVE
} else {
RecoveryState.RECOVERING
}
logInfo("I have been elected leader! New state: " + state)
if (state == RecoveryState.RECOVERING) {
beginRecovery(storedApps, storedDrivers, storedWorkers)
recoveryCompletionTask = forwardMessageThread.schedule(new Runnable {
override def run(): Unit = Utils.tryLogNonFatalError {
self.send(CompleteRecovery)
}
}, WORKER_TIMEOUT_MS, TimeUnit.MILLISECONDS)
}
}
/* 完成恢复 */
case CompleteRecovery => completeRecovery()
case RevokedLeadership => {
logError("Leadership has been revoked -- master shutting down.")
System.exit(0)
}
/* 注册worker */
case RegisterWorker(
id, workerHost, workerPort, workerRef, cores, memory, workerUiPort, publicAddress) => {
logInfo("Registering worker %s:%d with %d cores, %s RAM".format(
workerHost, workerPort, cores, Utils.megabytesToString(memory)))
/* 当状态为RecoveryState.STANDBY时，不注册 */
if (state == RecoveryState.STANDBY) {
// ignore, don't send response
} else if (idToWorker.contains(id)) {
/* 重复注册，通知注册失败 */
workerRef.send(RegisterWorkerFailed("Duplicate worker ID"))
} else {
val worker = new WorkerInfo(id, workerHost, workerPort, cores, memory,
workerRef, workerUiPort, publicAddress)
if (registerWorker(worker)) {
/* 注册成功，通知worker节点 */
persistenceEngine.addWorker(worker)
workerRef.send(RegisteredWorker(self, masterWebUiUrl))
schedule()
} else {
val workerAddress = worker.endpoint.address
logWarning("Worker registration failed. Attempted to re-register worker at same " +"address: " + workerAddress)
/* 注册失败，通知Worker节点 */
workerRef.send(RegisterWorkerFailed("Attempted to re-register worker at same address: "+ workerAddress))
}
}
}
/* 通知Executor的Driver更新状态 */
case ExecutorStateChanged(appId, execId, state, message, exitStatus) => {
……
override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {
case RequestSubmitDriver(description) => {
/* 当Master状态不为ALIVE的时候，通知Client无法提交Driver */
if (state != RecoveryState.ALIVE) {
val msg = s"${Utils.BACKUP_STANDALONE_MASTER_PREFIX}: $state. " +
"Can only accept driver submissions in ALIVE state."
context.reply(SubmitDriverResponse(self, false, None, msg))
} else {
logInfo("Driver submitted " + description.command.mainClass)
val driver = createDriver(description)
persistenceEngine.addDriver(driver)
waitingDrivers += driver
drivers.add(driver)
schedule()
/* 提交Driver */
context.reply(SubmitDriverResponse(self, true, Some(driver.id), s"Driver successfully submitted as ${driver.id}"))
}
}
case RequestKillDriver(driverId) => {
if (state != RecoveryState.ALIVE) {
val msg = s"${Utils.BACKUP_STANDALONE_MASTER_PREFIX}: $state. " + s"Can only kill drivers in ALIVE state."
/* 当Master不为ALIVE时，通知无法终止Driver */
context.reply(KillDriverResponse(self, driverId, success = false, msg))
} else {
logInfo("Asked to kill driver " + driverId)
val driver = drivers.find(_.id == driverId)
driver match {
case Some(d) =>
if (waitingDrivers.contains(d)) {
/* 当想kill的Driver在等待队列中时，删除Driver并更新状态为KILLED */
waitingDrivers -= d
self.send(DriverStateChanged(driverId, DriverState.KILLED, None))
} else {
/* 通知worker,Driver被终止 */
d.worker.foreach { w =>
w.endpoint.send(KillDriver(driverId))
}
}
// TODO: It would be nice for this to be a synchronous response
val msg = s"Kill request for $driverId submitted"
logInfo(msg)
/* 通知请求者，终止Driver的请求已提交 */
context.reply(KillDriverResponse(self, driverId, success = true, msg))
case None =>
val msg = s"Driver $driverId has already finished or does not exist"
logWarning(msg)
/* 通知请求者，Driver已被终止或不存在 */
context.reply(KillDriverResponse(self, driverId, success = false, msg))
}
}
}
……
3. Worker端通信逻辑
private[deploy] class Worker(
override val rpcEnv: RpcEnv,
webUiPort: Int,
cores: Int,
memory: Int,
masterRpcAddresses: Array[RpcAddress],
systemName: String,
endpointName: String,
workDirPath: String = null,
val conf: SparkConf,
val securityMgr: SecurityManager)
extends ThreadSafeRpcEndpoint with Logging {
……
override def receive: PartialFunction[Any, Unit] = {
/* 注册worker */
case RegisteredWorker(masterRef, masterWebUiUrl) =>
……
/* 向Master发送心跳 */
case SendHeartbeat =>
if (connected) { sendToMaster(Heartbeat(workerId, self)) }
/* 清理旧应用的工作目录 */
case WorkDirCleanup =>
// Spin up a separate thread (in a future) to do the dir cleanup; don't tie up worker
// rpcEndpoint.
// Copy ids so that it can be used in the cleanup thread.
val appIds = executors.values.map(_.appId).toSet
val cleanupFuture = concurrent.future {
……
/* 新Master选举产生时，Work更新Master相关信息，包括URL等 */
case MasterChanged(masterRef, masterWebUiUrl) =>
logInfo("Master has changed, new master is at " + masterRef.address.toSparkURL)
changeMaster(masterRef, masterWebUiUrl)
……
/* worker向主节点注册失败 */
case RegisterWorkerFailed(message) =>
if (!registered) {
logError("Worker registration failed: " + message)
System.exit(1)
}
/* worker重新连接向Master注册 */
case ReconnectWorker(masterUrl) =>
logInfo(s"Master with url $masterUrl requested this worker to reconnect.")
registerWithMaster()
/* 启动Executor */
case LaunchExecutor(masterUrl, appId, execId, appDesc, cores_, memory_) =>
……
/* 启动ExecutorRunner */
val manager = new ExecutorRunner(
……
/* executor状态改变 */
case executorStateChanged @ ExecutorStateChanged(appId, execId, state, message, exitStatus) =>
/* 通知Master executor状态改变 */
handleExecutorStateChanged(executorStateChanged)
/* 终止当前节点上运行的Executor */
case KillExecutor(masterUrl, appId, execId) =>
if (masterUrl != activeMasterUrl) {
logWarning("Invalid Master (" + masterUrl + ") attempted to launch executor " + execId)
} else {
val fullId = appId + "/" + execId
executors.get(fullId) match {
case Some(executor) =>
logInfo("Asked to kill executor " + fullId)
executor.kill()
case None =>
logInfo("Asked to kill unknown executor " + fullId)
}
……
/* 启动Driver */
case LaunchDriver(driverId, driverDesc) => {
logInfo(s"Asked to launch driver $driverId")
/* 创建DriverRunner */
val driver = new DriverRunner(...)
drivers(driverId) = driver
/* 启动Driver */
driver.start()
……
/* 终止worker节点上运行的Driver */
case KillDriver(driverId) => {
logInfo(s"Asked to kill driver $driverId")
drivers.get(driverId) match {
case Some(runner) =>
runner.kill()
case None =>
logError(s"Asked to kill unknown driver $driverId")
……
/* Driver状态更新 */
case driverStateChanged @ DriverStateChanged(driverId, state, exception) => {
handleDriverStateChanged(driverStateChanged)
}
……
3.5 容错机制及依赖
一般而言，对于分布式系统，数据集的容错性通常有两种方式：
1） 数据检查点（在Spark中对应Checkpoint机制）。
2） 记录数据的更新（在Spark中对应Lineage血统机制）。
对于大数据分析而言，数据检查点操作成本较高，需要通过数据中心的网络连接在机器之间复制庞大的数据集，而网络带宽往往比内存带宽低，同时会消耗大量存储资源。
Spark选择记录更新的方式。但更新粒度过细时，记录更新成本也不低。因此，RDD只支持粗粒度转换，即只记录单个块上执行的单个操作，然后将创建RDD的一系列变换序列记录下来，以便恢复丢失的分区。
3.5.1 Lineage（血统）机制
每个RDD除了包含分区信息外，还包含它从父辈RDD变换过来的步骤，以及如何重建某一块数据的信息，因此RDD的这种容错机制又称“血统”（Lineage）容错。Lineage本质上很类似于数据库中的重做日志（Redo Log），只不过这个重做日志粒度很大，是对全局数据做同样的重做以便恢复数据。
相比其他系统的细颗粒度的内存数据更新级别的备份或者LOG机制，RDD的Lineage记录的是粗颗粒度的特定数据Transformation操作（如filter、map、join等）。当这个RDD的部分分区数据丢失时，它可以通过Lineage获取足够的信息来重新计算和恢复丢失的数据分区。但这种数据模型粒度较粗，因此限制了Spark的应用场景。所以可以说Spark并不适用于所有高性能要求的场景，但同时相比细颗粒度的数据模型，也带来了性能方面的提升。
RDD在Lineage容错方面采用如下两种依赖来保证容错方面的性能：
窄依赖（Narrow Dependeny）：窄依赖是指父RDD的每一个分区最多被一个子RDD的分区所用，表现为一个父RDD的分区对应于一个子RDD的分区，或多个父RDD的分区对应于一个子RDD的分区。也就是说一个父RDD的一个分区不可能对应一个子RDD的多个分区。其中，1个父RDD分区对应1个子RDD分区，可以分为如下两种情况：
子RDD分区与父RDD分区一一对应（如map、filter等算子）。
一个子RDD分区对应N个父RDD分区（如co-paritioned（协同划分）过的Join）。
宽依赖（Wide Dependency，源码中称为Shuffle Dependency）：
宽依赖是指一个父RDD分区对应多个子RDD分区，可以分为如下两种情况：
一个父RDD对应所有子RDD分区（未经协同划分的Join）。
一个父RDD对应多个RDD分区（非全部分区）（如groupByKey）。
窄依赖与宽依赖关系如图3-10所示。
从图3-10可以看出对依赖类型的划分：根据父RDD分区是对应一个还是多个子RDD分区来区分窄依赖（父分区对应一个子分区）和宽依赖（父分区对应多个子分区）。如果对应多个，则当容错重算分区时，对于需要重新计算的子分区而言，只需要父分区的一部分数据，因此其余数据的重算就导致了冗余计算。
图3-10 两种依赖关系
对于宽依赖，Stage计算的输入和输出在不同的节点上，对于输入节点完好，而输出节点死机的情况，在通过重新计算恢复数据的情况下，这种方法容错是有效的，否则无效，因为无法重试，需要向上追溯其祖先看是否可以重试（这就是lineage，血统的意思），窄依赖对于数据的重算开销要远小于宽依赖的数据重算开销。
窄依赖和宽依赖的概念主要用在两个地方：一个是容错中相当于Redo日志的功能；另一个是在调度中构建DAG作为不同Stage的划分点（前面调度机制中已讲过）。
依赖关系在lineage容错中的应用总结如下：
1）窄依赖可以在某个计算节点上直接通过计算父RDD的某块数据计算得到子RDD对应的某块数据；宽依赖则要等到父RDD所有数据都计算完成，并且父RDD的计算结果进行hash并传到对应节点上之后，才能计算子RDD。
2）数据丢失时，对于窄依赖，只需要重新计算丢失的那一块数据来恢复；对于宽依赖，则要将祖先RDD中的所有数据块全部重新计算来恢复。所以在长“血统”链特别是有宽依赖时，需要在适当的时机设置数据检查点（checkpoint机制在下节讲述）。可见Spark在容错性方面要求对于不同依赖关系要采取不同的任务调度机制和容错恢复机制。
在Spark容错机制中，如果一个节点宕机了，而且运算属于窄依赖，则只要重算丢失的父RDD分区即可，不依赖于其他节点。而宽依赖需要父RDD的所有分区都存在，重算就很昂贵了。更深入地来说：在窄依赖关系中，当子RDD的分区丢失，重算其父RDD分区时，父RDD相应分区的所有数据都是子RDD分区的数据，因此不存在冗余计算。而在宽依赖情况下，丢失一个子RDD分区重算的每个父RDD的每个分区的所有数据并不是都给丢失的子RDD分区使用，其中有一部分数据对应的是其他不需要重新计算的子RDD分区中的数据，因此在宽依赖关系下，这样计算就会产生冗余开销，这也是宽依赖开销更大的原因。为了减少这种冗余开销，通常在Lineage血统链比较长，并且含有宽依赖关系的容错中使用Checkpoint机制设置检查点。
3.5.2 Checkpoint（检查点）机制
通过上述分析可以看出Checkpoint的本质是将RDD写入Disk来作为检查点。这种做法是为了通过lineage血统做容错的辅助，lineage过长会造成容错成本过高，这样就不如在中间阶段做检查点容错，如果之后有节点出现问题而丢失分区，从做检查点的RDD开始重做Lineage，就会减少开销。
下面从代码层面介绍Checkpoint的实现。
1. 设置检查点数据的存取路径[SparkContext.scala]
/* 设置作为RDD检查点的目录，如果是集群上运行，则必须为HDFS路径 */
def setCheckpointDir(directory: String) {
// If we are running on a cluster, log a warning if the directory is local.
// Otherwise, the driver may attempt to reconstruct the checkpointed RDD from
// its own local file system, which is incorrect because the checkpoint files
// are actually on the executor machines.
if (!isLocal && Utils.nonLocalPaths(directory).isEmpty) {
logWarning("Checkpoint directory must be non-local " +
"if Spark is running on a cluster: " + directory)
}
checkpointDir = Option(directory).map { dir =>
val path = new Path(dir, UUID.randomUUID().toString)
val fs = path.getFileSystem(hadoopConfiguration)
fs.mkdirs(path)
fs.getFileStatus(path).getPath.toString
}
}
2. 设置检查点的具体实现
[RDD.scala]
/* 设置检查点入口 */
private[spark] def doCheckpoint(): Unit = {
RDDOperationScope.withScope(sc, "checkpoint", allowNesting = false, ignoreParent = true) {
if (!doCheckpointCalled) {
doCheckpointCalled = true
if (checkpointData.isDefined) {
checkpointData.get.checkpoint()
} else {
/* */
dependencies.foreach(_.rdd.doCheckpoint())
}
}
}
}
[RDDCheckPointData.scala]
/* 设置检查点，在子类中会覆盖此函数以实现具体功能 */
protected def doCheckpoint(): CheckpointRDD[T]
[ReliableRDDCheckpointData.scala]
/* 设置检查点，将RDD内容写入可靠的分布式文件系统中 */
protected override def doCheckpoint(): CheckpointRDD[T] = {
/* 为检查点创建输出目录 */
val path = new Path(cpDir)
val fs = path.getFileSystem(rdd.context.hadoopConfiguration)
if (!fs.mkdirs(path)) {
throw new SparkException(s"Failed to create checkpoint path $cpDir")
}
/* 保存为文件，加载时作为一个RDD加载 */
val broadcastedConf = rdd.context.broadcast(
new SerializableConfiguration(rdd.context.hadoopConfiguration))
/* 重新计算RDD */
rdd.context.runJob(rdd, ReliableCheckpointRDD.writeCheckpointFile[T](cpDir, broadcastedConf) _)
val newRDD = new ReliableCheckpointRDD[T](rdd.context, cpDir)
if (newRDD.partitions.length != rdd.partitions.length) {
throw new SparkException(
s"Checkpoint RDD $newRDD(${newRDD.partitions.length}) has different " +
s"number of partitions from original RDD $rdd(${rdd.partitions.length})")
}
/* 当引用不在此范围时，清除检查点文件 */
if (rdd.conf.getBoolean("spark.cleaner.referenceTracking.cleanCheckpoints", false)) {
rdd.context.cleaner.foreach { cleaner =>
cleaner.registerRDDCheckpointDataForCleanup(newRDD, rdd.id)
}
}
logInfo(s"Done checkpointing RDD ${rdd.id} to $cpDir, new parent is RDD ${newRDD.id}")
newRDD
}
}
3.6 Shuffle机制
在MapReduce框架中，Shuffle是连接Map和Reduce之间的桥梁，Map的输出要用到Reduce中必须经过Shuffle这个环节，Shuffle的性能高低直接影响了整个程序的性能和吞吐量。Spark作为MapReduce框架的一种实现，自然也实现了Shuffle的逻辑。对于大数据计算框架而言，Shuffle阶段的效率是决定性能好坏的关键因素之一。
3.6.1 什么是Shuffle
Shuffle是MapReduce框架中的一个特定的阶段，介于Map阶段和Reduce阶段之间，当Map的输出结果要被Reduce使用时，输出结果需要按关键字值（key）哈希，并且分发到每一个Reducer上，这个过程就是Shuffle。直观来讲，Spark Shuffle机制是将一组无规则的数据转换为一组具有一定规则数据的过程。由于Shuffle涉及了磁盘的读写和网络的传输，因此Shuffle性能的高低直接影响整个程序的运行效率。
在MapReduce计算框架中，Shuffle连接了Map阶段和Reduce阶段，即每个Reduce Task从每个Map Task产生的数据中读取一片数据，极限情况下可能触发M*R个数据拷贝通道（M是Map Task数目，R是Reduce Task数目）。通常Shuffle分为两部分：Map阶段的数据准备和Reduce阶段的数据拷贝。首先，Map阶段需根据Reduce阶段的Task数量决定每个Map Task输出的数据分片数目，有多种方式存放这些数据分片：
1） 保存在内存中或者磁盘上（Spark和MapReduce都存放在磁盘上）。
2） 每个分片对应一个文件（现在Spark采用的方式，以及以前MapReduce采用的方式），或者所有分片放到一个数据文件中，外加一个索引文件记录每个分片在数据文件中的偏移量（现在MapReduce采用的方式）。
因此可以认为Spark Shuffle与Mapreduce Shuffle的设计思想相同，但在实现细节和优化方式上不同。
在Spark中，任务通常分为两种，Shuffle mapTask和reduceTask，具体逻辑如图3-11所示：
图3-11 Spark Shuffle
图3-11中的主要逻辑如下：
1）首先每一个MapTask会根据ReduceTask的数量创建出相应的bucket，bucket的数量是M×R，其中M是Map的个数，R是Reduce的个数。
2）其次MapTask产生的结果会根据设置的partition算法填充到每个bucket中。这里的partition算法是可以自定义的，当然默认的算法是根据key哈希到不同的bucket中。
当ReduceTask启动时，它会根据自己task的id和所依赖的Mapper的id从远端或本地的block manager中取得相应的bucket作为Reducer的输入进行处理。
这里的bucket是一个抽象概念，在实现中每个bucket可以对应一个文件，可以对应文件的一部分或是其他等。Spark shuffle可以分为两部分：
1） 将数据分成bucket，并将其写入磁盘的过程称为Shuffle Write。
2） 在存储Shuffle数据的节点Fetch数据，并执行用户定义的聚集操作，这个过程称为Shuffle Fetch。
3.6.2 Shuffle历史及细节
下面介绍Shuffle Write与Fetch。
1. Shuffle Write
在Spark的早期版本实现中，Spark在每一个MapTask中为每个ReduceTask创建一个bucket，并将RDD计算结果放进bucket中。
但早期的Shuffle Write有两个比较大的问题。
1）Map的输出必须先全部存储到内存中，然后写入磁盘。这对内存是非常大的开销，当内存不足以存储所有的Map输出时就会出现OOM（Out of Memory）。
2）每个MapTask会产生与ReduceTask数量一致的Shuffle文件，如果MapTask个数是1k，ReduceTask个数也是1k，就会产生1M个Shuffle文件。这对于文件系统是比较大的压力，同时在Shuffle数据量不大而Shuffle文件又非常多的情况下，随机写也会严重降低IO的性能。
后来到了Spark 0.8版实现时，显著减少了Shuffle的内存压力，现在Map输出不需要先全部存储在内存中，再flush到硬盘，而是record-by-record写入磁盘中。对于Shuffle文件的管理也独立出新的ShuffleBlockManager进行管理，而不是与RDD cache文件在一起了。
但是Spark 0.8版的Shuffle Write仍然有两个大的问题没有解决。
1）Shuffle文件过多的问题。这会导致文件系统的压力过大并降低IO的吞吐量。
2）虽然Map输出数据不再需要预先存储在内存中然后写入磁盘，从而显著减少了内存压力。但是新引入的DiskObjectWriter所带来的buffer开销也是不容小视的内存开销。假定有1k个MapTask和1k个ReduceTask，就会有1M个bucket，相应地就会有1M个write handler，而每一个write handler默认需要100KB内存，那么总共需要100GB内存。这样仅仅是buffer就需要这么多的内存。因此当ReduceTask数量很多时，内存开销会很大。
为了解决shuffle文件过多的情况，Spark后来引入了新的Shuffle consolidation，以期显著减少Shuffle文件的数量。
Shuffle consolidation的原理如图3-12所示：
在图3-12中，假定该job有4个Mapper和4个Reducer，有2个core能并行运行两个task。可以算出Spark的Shuffle Write共需要16个bucket，也就有了16个write handler。在之前的Spark版本中，每个bucket对应一个文件，因此在这里会产生16个shuffle文件。
图3-12 Shuffle consolidation
而在Shuffle consolidation中，每个bucket并非对应一个文件，而是对应文件中的一个segment。同时Shuffle consolidation产生的Shuffle文件数量与Spark core的个数也有关系。在图3-12中，job中的4个Mapper分为两批运行，在第一批2个Mapper运行时会申请8个bucket，产生8个Shuffle文件；而在第二批Mapper运行时，申请的8个bucket并不会再产生8个新的文件，而是追加写到之前的8个文件后面，这样一共就只有8个Shuffle文件，而在文件内部共有16个不同的segment。因此从理论上讲Shuffle consolidation产生的Shuffle文件数量为C×R，其中C是Spark集群的core number，R是Reducer的个数。
很显然，当M=C时，Shuffle consolidation产生的文件数和之前的实现相同。
Shuffle consolidation显著减少了Shuffle文件的数量，解决了Spark之前实现中一个比较严重的问题。但是Writer handler的buffer开销过大依然没有减少，若要减少Writer handler的buffer开销，只能减少Reducer的数量，但是这又会引入新的问题。
2. Shuffle Fetch与Aggregator
Shuffle Write写出去的数据要被Reducer使用，就需要Shuffle Fetch将所需的数据Fetch过来。这里的Fetch操作包括本地和远端，因为Shuffle数据有可能一部分是存储在本地的。在早期版本中，Spark对Shuffle Fetcher实现了两套不同的框架：NIO通过socket连接Fetch数据；OIO通过netty server去fetch数据。分别对应的类是Basic-BlockFetcherIterator和NettyBlockFetcherIterator。
目前在Spark1.5.0中做了优化。新版本定义了类ShuffleBlockFetcherIterator来完成数据的fetch。对于local的数据，ShuffleBlockFetcherIterator会通过local的BlockMan-ager来fetch。对于远端的数据块，它通过BlockTransferService类来完成。具体实现参见如下代码：
[ShuffleBlockFetcherIterator.scala]
/* fetch local数据块 */
private[this] def fetchLocalBlocks() {
val iter = localBlocks.iterator
while (iter.hasNext) {
val blockId = iter.next()
try {
/* 通过blockManager来fetch数据 */
val buf = blockManager.getBlockData(blockId)
shuffleMetrics.incLocalBlocksFetched(1)
shuffleMetrics.incLocalBytesRead(buf.size)
buf.retain()
results.put(new SuccessFetchResult(blockId, blockManager.blockManagerId, 0, buf))
} catch {
case e: Exception =>
// If we see an exception, stop immediately.
logError(s"Error occurred while fetching local blocks", e)
results.put(new FailureFetchResult(blockId, blockManager.blockManagerId, e))
return
}
}
}
/* 发送请求获取远端数据 */
private[this] def sendRequest(req: FetchRequest) {
/* 请求格式 */
logDebug("Sending request for %d blocks (%s) from %s".format(
req.blocks.size, Utils.bytesToString(req.size), req.address.hostPort))
bytesInFlight += req.size
// so we can look up the size of each blockID
val sizeMap = req.blocks.map { case (blockId, size) => (blockId.toString, size) }.toMap
val blockIds = req.blocks.map(_._1.toString)
val address = req.address
/* fetch数据 */
shuffleClient.fetchBlocks(address.host, address.port, address.executorId, blockIds.toArray,
new BlockFetchingListener {
override def onBlockFetchSuccess(blockId: String, buf: ManagedBuffer): Unit = {
// Only add the buffer to results queue if the iterator is not zombie,
// i.e. cleanup() has not been called yet.
if (!isZombie) {
// Increment the ref count because we need to pass this to a different thread.
// This needs to be released after use.
buf.retain()
/* fetch请求成功 */
results.put(new SuccessFetchResult(BlockId(blockId), address, sizeMap(blockId), buf))
shuffleMetrics.incRemoteBytesRead(buf.size)
shuffleMetrics.incRemoteBlocksFetched(1)
}
……
}
override def onBlockFetchFailure(blockId: String, e: Throwable):
/* fetch 失败*/
……
}
}
在MapReduce的Shuffle过程中，Shuffle fetch过来的数据会进行归并排序（merge sort），使得相同key下的不同value按序归并到一起供Reducer使用，这个过程如图3-13所示：
这些归并排序都是在磁盘上进行的，这样做虽然有效地控制了内存使用，但磁盘IO却大幅增加了。虽然Spark属于MapReduce体系，但是对传统的MapReduce算法进行了一定的改变。Spark假定在大多数应用场景下，Shuffle数据的排序不是必须的，如word count。强制进行排序只会使性能变差，因此Spark并不在Reducer端做归并排序。既然没有归并排序，那Spark是如何进行reduce的呢？这就涉及下面要讲的Shuffle Aggregator了。
图3-13 Fetch merge
Aggregator本质上是一个hashmap，它是以map output的key为key，以任意所要combine的类型为value的hashmap。
在做word count reduce计算count值时，它会将Shuffle fetch到的每一个key-value对更新或是插入hashmap中（若在hashmap中没有查找到，则插入其中；若查找到，则更新value值）。这样就不需要预先把所有的key-value进行merge sort，而是来一个处理一个，省去了外部排序这一步骤。但同时需要注意的是，reducer的内存必须足以存放这个partition的所有key和count值，因此对内存有一定的要求。
在上面word count的例子中，因为value会不断地更新，而不需要将其全部记录在内存中，因此内存的使用还是比较少的。考虑一下如果是groupByKey这样的操作，Reducer需要得到key对应的所有value。在Hadoop MapReduce中，由于有了归并排序，因此给予Reducer的数据已经是group by key了，而Spark没有这一步，因此需要将key和对应的value全部存放在hashmap中，并将value合并成一个array。可以想象为了能够存放所有数据，用户必须确保每一个partition小到内存能够容纳，这对于内存是非常严峻的考验。因此在Spark文档中，建议用户涉及这类操作时尽量增加partition，也就是增加Mapper和Reducer的数量。
增加Mapper和Reducer的数量固然可以减小partition的大小，使内存可以容纳这个partition。但是在Shuffle write中提到，bucket和对应于bucket的write handler是由Mapper和Reducer的数量决定的，task越多，bucket就会增加得更多，由此带来write handler所需的buffer也会更多。在一方面我们为了减少内存的使用采取了增加task数量的策略，另一方面task数量增多又会带来buffer开销更大的问题，因此陷入了内存使用的两难境地。
为了减少内存的使用，只能将Aggregator的操作从内存移到磁盘上进行，因此Spark新版本中提供了外部排序的实现，以解决这个问题。
Spark将需要聚集的数据分为两类：不需要归并排序和需要归并排序的数据。对于前者，在内存中的AppendOnlyMap中对数据聚集。对于需要归并排序的数据，现在内存中进行聚集，当内存数据达到阈值时，将数据排序后写入磁盘。事实上，磁盘上的数据只是全部数据的一部分，最后将磁盘数据全部进行归并排序和聚集。具体Aggregator的逻辑可以参见Aggregator类的实现。
@DeveloperApi
case class Aggregator[K, V, C] (
createCombiner: V => C,
mergeValue: (C, V) => C,
mergeCombiners: (C, C) => C) {
// 是否外部排序
private val isSpillEnabled = SparkEnv.get.conf.getBoolean("spark.shuffle.spill", true)
@deprecated("use combineValuesByKey with TaskContext argument", "0.9.0")
def combineValuesByKey(iter: Iterator[_ <: Product2[K, V]]): Iterator[(K, C)] =
combineValuesByKey(iter, null)
def combineValuesByKey(iter: Iterator[_ <: Product2[K, V]],
context: TaskContext): Iterator[(K, C)] = {
if (!isSpillEnabled) {
/* 创建AppendOnlyMap对象存储了combine集合，每个combine是一个Key及对应Key的元素Seq */
val combiners = new AppendOnlyMap[K, C]
var kv: Product2[K, V] = null
val update = (hadValue: Boolean, oldValue: C) => {
/* 检查是否处理的是第一个元素，如果是则先创建集合结构，如果不是则直接插入 */
if (hadValue) mergeValue(oldValue, kv._2) else createCombiner(kv._2)
}
while (iter.hasNext) {
kv = iter.next()
/* 当不采用外排时，利用AppendOnlyMap结构存储数据 */
combiners.changeValue(kv._1, update)
}
combiners.iterator
} else {
val combiners = new ExternalAppendOnlyMap[K, V, C](createCombiner, mergeValue, mergeCombiners)
/* 如果采用外排时，使用ExternalAppendOnlyMap结构存储聚集数据 */
combiners.insertAll(iter)
updateMetrics(context, combiners)
combiners.iterator
……
本节就Shuffle的概念与原理先介绍到这里。在下一章讲解Spark源码时，会对Shuffle的核心机制——Shuffle存储做代码层面的讲解。相信学习完本章和第4章的Shuffle存储机制后，读者会对Shuffle机制掌握得更加深入。
3.7 本章小结
本章主要讲述了Spark的工作机制与原理。首先剖析了Spark的提交和执行时的具体机制，重点强调了Spark程序的宏观执行过程： 提交后的Job在Spark中形成了RDD DAG（有向无环图），然后进入一系列切分调度的过程。在剖析过程中，结合Spark的源码呈现了这些调度过程的代码细节。本章后半部分接着剖析了Spark的存储及IO、Spark通信机制，最后讲述了Spark的容错机制及Shuffle机制。 本章内容比较多，希望读者仔细体会。
