Spark原理篇之工作原理

1 Spark背景

Spark是一个加州大学伯克利分校（UC Berkeley AMP）开发的一个分布式数据快速分析项目。它的核心技术是弹性分布式数据集（Resilient distributed datasets），提供了比Hadoop更加丰富的MapReduce模型，可以快速在内存中对数据集进行多次迭代，来支持复杂的数据挖掘算法和图计算算法。

2 Spark工作原理

2.1 Spark专业术语定义

（1）Application：Spark应用程序
      指的是用户编写的Spark应用程序，包含Driver功能代码和分布在集群中多个节点上运行的Executor代码。Spark应用程序，由一个或多个作业Job组成，如下图所示：
（2）Driver：驱动程序
      Spark中的Driver即运行上述Application的Main()函数并且创建SparkContext，其中创建SparkContext的目的是为了准备Spark应用程序的运行环境。在Spark中由SparkContext负责和ClusterManager通信，进行资源的申请、任务的分配和监控等；当Executor部分运行完毕后，Driver负责将SparkContext关闭。通常SparkContext代表Driver，如下图所示：

（3）Cluster Manager：资源管理器
      指的是在集群上获取资源的外部服务，常用的有：Local，Spark单机运行，一般用于开发测试；Standalone，Spark原生的资源管理器，由Master负责资源的分配；Hadoop Yarn，由Yarn中的ResearchManager负责资源的分配；由Messos中的Messos Master负责资源管理。
（4）Executor：执行器
      Application运行在Worker节点上的一个进程，该进程负责运行Task，并且负责将数据存在内存或者磁盘上，每个Application都有各自独立的一批Executor，如下图所示：

（5）Worker：计算节点
      集群中任何可以运行Application代码的节点，类似于Yarn中的NodeManager节点。在Standalone模式中指的是通过Slave文件配置的Worker节点，在Spark on Yarn模式汇总指的就是NodeManager节点，在Spark on Messos模式中指的就是Messos Slave节点，如下图所示：

（6）RDD (Resilient distributed datasets)：弹性分布式数据集
      Spark的基本计算单元，可以通过一系列算子进行操作（主要有Transformation和Action操作），如下图所示：

（7）窄依赖
      父RDD每一个分区最多被一个子RDD的分区所用；表现为一个父RDD的分区对应于一个子RDD的分区，或两个父RDD的分区对应于一个子RDD的分区。如图所示：
      一个Stage内的窄依赖进行pipeline操作（Spark为了加快计算，做的优化），举例：
      ① 优化：1+1+1+1=4；
      ② 未优化：1+1=2，2+1=3，3+1=4。
（8）宽依赖
      父RDD的每个分区都可能被多个子RDD分区所使用，子RDD分区通常对应所有的父 RDD分区。如图所示：
      常见的宽窄依赖有：map、filter、union、mapPartitions、mapValues、join（父RDD是hash-partitioned：如果joinAPI之前被调用的RDD API是宽依赖【存在shuffle】，而且两个join的RDD的分区数量一致，join结果的RDD分区数量也一样，这个时候join API是窄依赖）。
      常见的宽依赖有：groupByKey、partitionBy、reduceByKey、join（父RDD不是hash-partitioned：除此之外的，RDD的join API都是宽依赖）。
      对hash-partitioned想要更详细的理解，请参考文章[4][5]。
（9）DAG (Directed Acycle Graph)：有向无环图
      反应RDD之间的依赖关系，如图所示：
（10）DAGScheduler：有向无环图调度器
      基于DAG划分Stage并以TaskSet的形式提交Stage给TaskScheduler；负责将作业拆分成不同阶段的具有宽窄依赖关系的多批任务；最重要的任务之一就是：计算作业和任务的依赖关系，指定调度逻辑。在SparkContext初始化的过程中被实例化，一个SparkContext对应一个DAGScheduler。
（11）TaskScheduler：任务调度器
      将TaskSet提交给Worker（集群）运行并汇报结果；负责每个具体任务的实际物理调度。如图所示：
（12）Job：作业
      由一个或多个调度阶段所组成的一次计算作业；包含多个Task组成的并行计算，往往由Spark Action催生，一个Job包含多个RDD及作用于相应RDD上的各种Operation。如图所示：
（13）Stage：调度阶段
      一个任务对应的阶段；每个Job会被拆分很多组Task，每组任务被称为Stage，也可称TaskSet，一个作业分为多个阶段；Satge分成两种类型ShuffleMapStage、ResultStage。如图所示：
（14）TaskSet：任务集
      由一组关联的，但相互之间没有Shuffle依赖关系的任务所组成的任务集。如图所示：
提示：
      1）一个Stage创建一个TaskSet；
      2）为Stage的每个RDD分区创建一个Task，多个Task分装成TaskSet。
（15）Task：任务
      被送到某个Executor上的工作任务；单个分区数据集上的最小处理流程单元。如图所示：

      总体如图所示：

2.2 Spark基本流程

（1）Spark内核架构图

1-14步属于SparkContext的初始化阶段。
（2）运行基本流程

2.3 Spark运行架构特点

（1）Executor进程专属
      每个Application获取专属的Executor进程，该进程在Application期间一直驻留，并以多线程方式运行Tasks。Spark Application不能跨应用程序共享数据，除非将数据写入到外部存储系统。如图所示：
（2）支持多种资源管理器
      Spark与资源管理器无关，只要能够获取Executor进程，并能保持相互通信就可以了，Spark支持资源管理器（Standalone、Messos和Yarn）
（3）Job提交就近原则
      提交SparkContext的Client靠近Worker节点（运行Executor的节点），最好是在同一个Rack（机架）里，因为Spark Application运行过程中SparkContext和Executor之间有大量的信息交换；如果想在远程集群中运行，最好使用RPC将SparkContext提交给集群，不要远离Worker运行SparkContext。
（4）移动程序而非移动数据的原则执行
      Task采用了数据本地性和推测执行的优化机制。关键方法：taskIdToLocations、getPreferedLocations。
2.4 Spark核心原理透视
（1）计算流程
（2）从代码构建DAG图
      Spark的计算发生在RDD的Action操作，而对Action之前的所有Transformation，Spark只是记录下RDD生成的轨迹，而不会触发真正的计算。
Spark内核会在需要计算发生的时刻绘制一张关于计算路径的有向无环图，也就是DAG。

（3）将DAG划分为Stage核心算法
      Application多个Job多个Stage：Spark Application中可以因为不同的Action触发众多的Job，一个Application中可以有很多的Job，每个Job是由一个或多个Stage构成的，后面的Stage依赖于前面的Stage，也就是说只有前面的Stage计算完毕后，后面的Stage才会运行。
      划分依据：Stage划分的依据就是宽依赖，何时产生宽依赖，reduceByKey，groupByKey等算子，会导致宽依赖的产生。
      核心算法：从后往前回溯，遇到窄依赖加入本Stage，遇见宽依赖进行Stage切分。Spark内核会从触发Action操作的那个RDD开始从后往前推，首先会为最后一个RDD创建一个Stage，然后继续倒推，如果发现对某个RDD是宽依赖，那么就会将宽依赖的那个RDD创建一个新的Stage，那个RDD就是新的Stage的最后一个RDD。然后以此类推，继续倒推，根据窄依赖或者宽依赖进行Stage的划分，直到所有的RDD全部遍历完成为止。
（4）将DAG划分为Stage剖析
      从HDFS中读取数据生成3个不同的RDD，通过一系列Transformation操作后再将计算结果保存到HDFS。可以看到这个DAG中只有join操作是一个窄依赖，Spark内核会以此为边界将其前后划分成不同的Stage。同时我们可以注意到，在图中Stage2中，从map到union都是窄依赖，这两步操作可以形成一个流水线操作，通过map操作生成的partition可以不用等待整个RDD计算结束，而是继续进行union操作，这样大大提高了计算的效率。
（5）提交Stages
      调度阶段的提交，最终会将转换成一个任务集的提交，DAGScheduler通过TaskScheduler接口提交任务集，这个任务集最终会触发TaskScheduler构建一个TaskSetManager的实例来管理这个任务集的生命周期，对于DAGScheduler来说，提交调度阶段的工作到此就完成了。而TaskScheduler的具体实现则会在得到计算资源的时候，进一步通过TaskSetManager调度具体的任务到对应的Executor节点上进行运算。

      TaskSetManager负责管理TaskSchedulerImpl中一个单独TaskSet，跟踪每一个Task，如果Task失败，负责重试Task直到达到Task重试次数的最多次数。
（6）监控Job、Task和Executor
      DAGScheduler监控Job和Task：要保证相互依赖的作业调度阶段能够得到顺利地调度执行。DAGScheduler需要监控当前作业调度阶段乃至任务的完成情况。这通过对外暴露的一系列的回调函数来实现的，对于TaskScheduler来说，这些回调函数主要包括任务的开始结束失败、任务集的失败，DAGScheduler根据这些任务的生命周期信息进一步维护作业和调度阶段的状态信息。
      DAGScheduler监控Executor的生命状态：TaskScheduler通过回调函数通知DAGScheduler具体的Executor的生命状态，如果某一个Executor崩溃了，则对应的调度阶段任务集的ShuffleMap Task的输出结果也将标志位不可用，这将导致对应任务集状态的变更，进而重新执行相关计算任务，以获取丢失的相关数据。
（7）获取任务执行结果
      结果DAGScheduler：一个具体的任务在Executor中执行完毕后，其结果需要以某种形式返回给DAGScheduler，根据任务类型的不同，任务结果的返回方式也不同。
      两种结果，中间结果与最终结果：对于FinalStage所对应的任务，返回给DAGScheduler的是运算结果本身，而对于中间阶段对应的任务ShuffleMap Task，返回给DAGScheduler的是一个MapStatus里的相关存储信息，而非结果本身，这些存储位置信息作为下一个调度阶段的任务获取输入数据的依据。
      两种类型，DirectTaskResult与IndirectTaskResult：根据任务结果大小的不同，ResultTask返回的结果又分为两类，如果结果足够小，则直接放在DirectTaskResult对象内存中，如果超过特定尺寸则在Executor端会将DirectTaskResult先序列化，再把序列化的结果作为一个数据块放在BlockManager中，然后将BlockManager返回的BlockID放在IndirectTaskResult对象中返回给TaskScheduler，TaskScheduler进而调用TaskResultGetter将IndirectTaskResult中的BlockID取出并通过BlockManager最终取得对应的DirectTaskResult。
（8）任务调度总体诠释

2.4 Spark运行模式

2.4.1 Standalone模式

（1）Standalone架构图
（2）Standalone运行过程
      ① SparkContext连接到Master，向Master注册并申请资源（CPU Core and Memory）。
      ② Master根据SparkContext的资源申请要求和Worker心跳周期内报告的信息决定哪个Worker上分配资源，然后在该Worker上获取资源，然后启动Executor，Executor向SparkContext注册。
      ③ SparkContext将Application代码发送到Executor。同时，SparkContext解析Application代码，构建DAG图，并提交给DAGScheduler分解成Stage，然后以Stage（或者称为TaskSet）提交给TaskScheduler，TaskScheduler负责将Task分配到相应的Worker，最后提交给Executor执行。
      ④ Executor会建立Executor线程池，开始执行Task，并向SparkContext报告，直至Task完成。
      ⑤ 所有Task完成后，SparkContext向Master注销，释放资源。如图所示：

2.4.2 Spark on Yarn模式

（1）Yarn-Cluster模式
      ① Client向Yarn中提交应用程序，包括ApplicationMaster程序、启动ApplicationMaster的命令、需要在Executor中运行的程序等。
      ② ResourceManager收到请求后，在集群中选择一个NodeManager，为该应用程序分配第一个Container，要求它在这个Container中启动应用程序的ApplicationMaster，其中ApplicationMaster进行SparkContext等的初始化。
      ③ ApplicationMaster向ResourceManager注册，这样用户可以直接通过ResourceManage查看应用程序的运行状态，然后它将采用轮询的方式通过RPC协议为各个任务申请资源，并监控它们的运行状态直到运行结束。
      ④ 一旦ApplicationMaster申请到资源后，便与对应的NodeManager通信，要求它在获得的Container中启动Executor，启动后会向ApplicationMaster中的SparkContext注册并申请Task。
      ⑤ ApplicationMaster中的SparkContext分配Task给Executor执行，Executor运行Task并向ApplicationMaster汇报运行的状态和进度，以让ApplicationMaster随时掌握各个任务的运行状态，从而可以在任务失败时重新启动任务。
      ⑥ 应用程序运行完成后，ApplicationMaster向ResourceManager申请注销并关闭自己。
（2）Spark on Yarn-Client 模式
      ① Client向Yarn的ResourceManager申请启动ApplicationMaster。同时，在SparkContext初始化中将创建DAGScheduler和TaskScheduler。
      ② ResourceManager收到请求后，在集群中选择一个NodeManager，为该能够用程序分配第一个Container，要求它在这个Container中启动应用程序的ApplicationMaster。
      ③ Client中的SparkContext初始化完毕后，与ApplicationMaster建立通讯，向ResourceManager注册，根据任务信息向ResourceManager申请资源。
      ④ 一旦ApplicationMaster申请到资源后，便与对应的NodeManager通信，要求它在获得的Container中启动Executor，启动后会向Client中的SparkContext注册并申请Task。
      ⑤ Client中的SparkContext分配Task给Executor执行，Executor运行Task并向Driver汇报运行的状态和进度，以让Client随时掌握各个任务的运行状态，从而可以在任务失败时重新启动任务。
      ⑥ 应用程序运行完成后，Client的SparkContext向ResourceManager申请注销并关闭自己。
（3）Yarn-Cluster与Yarn-Client的区别
      它们的区别就是ApplicationMaster的区别：Yarn-Cluster中ApplicationMaster不仅负责申请资源，并负责监控Task的运行状况，因此可以关掉Client；而Yarn-Client中ApplicationMaster仅负责申请资源，由Client中的Driver来监控调度Task的运行，因此不能关掉Client。
（4）Spark on Yarn与MapReduce on Yarn的区别
      Spark的高性能一定程度上取决于它采用的异步并发模型（这里指server/driver端采用的模型），这与Hadoop 2.0（包括Yarn和MapReduce）是一致的。Hadoop 2.0自己实现了类似Actor的异步并发模型，实现方式是epoll+状态机，而Spark则直接采用了开源软件Akka，该软件实现了Actor模型，性能非常高。尽管二者在server端采用了一致的并发模型，但在任务级别（特指Spark任务和MapReduce任务）上却采用了不同的并发机制：Hadoop MapReduce采用了多进程模型，而Spark采用了多线程模型。
      注意，这里的多进程和多线程，指的是同一个节点上多个任务的运行模式。无论是MapReduce和Spark，整体上看，都是多进程：MapReduce应用程序是由多个独立的Task进程组成的；Spark应用成的运行环境是由多个独立的Executor进程构建的临时资源池构成的。
      多进程模型便于细粒度控制每个任务占用的资源，但会消耗较多的启动时间，不适合运行低延迟类型的作业，这是MapReduce广为诟病的原因之一。而多线程模型则相反，该模型使得Spark很适合运行低延迟类型的作业。总之，Spark同节点上的任务以多线程的方式运行在一个JVM进程中，可带来以下好处：
      ① 任务启动速度快，与之相反的是MapReduce Task进程的慢启动速度，通常需要1s左右；
      ② 同节点上所有任务运行在一个进程中，有利于共享内存。这非常适合内存密集型任务，尤其对于那些需要加载大量词典的应用程序，可大大节省内存；
      ③ 同节点上所有任务可运行在一个JVM进程（Executor）中，且Executor所占资源可连续被多批任务使用，不会在运行部分任务后释放掉，这避免了每个任务重复申请资源带来的时间开销，对于任务数目非常多的应用，可大大降低运行时间。与之对比的是MapReduce中的Task：每个Task单独申请资源，用完后马上释放，不能被其他任务重用，尽管1.0支持JVM重用在一定程度上弥补了该问题，但2.0尚未支持该功能。
      尽管Spark的多线程模型带来了很多好处，但同样存在不足，主要有：
      ① 由于节点上所有任务运行在一个进程中，因此，会出现严重的资源争用，难以细粒度控制每个任务占用资源。与之相反的是MapReduce，它允许用户单独为Map Task和Reduce Task设置不同的资源，进而细粒度控制任务占用资源量，有利于大作业的正常平稳运行。
      下面简要介绍MapReduce的多进程模型进而Spark多线程模型
      1）MapReduce多进程模型
      ① 每个Task运行在一个独立的JVM进程中；
      ② 可单独为不同类型的Task设置不同的资源量，目前支持内存和CPU两种资源；
      ③ 每个Task运行完后，释放所占用的资源，这些资源不能被其他Task复用，即使是同一个作业相同类型的Task。也就是说，每个Task都要经历“申请资源->运行Task->释放资源”的过程。
2）Spark多线程模型
      ① 每个节点上可以运行一个或多个Executor服务（一个节点上可以有多个Container）；
      ② 每个Executor配有一定数量的slot，表示该Executor中可以同时运行多少个ShuffleMap Task或者Reduce Task；
      ③ 每个Executor单独运行在一个JVM进程中，每个Task则是运行在Executor中的一个线程（每个线程可以通过轮询的方式运行每个Task）；
      ④ 同一个Executor内部的Task可共享内存，比如通过函数SparkContext.broadcast广播的文件或者数据结构只会在每个Executor中加载一次，而不会像MapReduce那样，每个Task加载一次；
      ⑤ Executor一旦启动后，将一直运行，且它的资源可以一直被Task复用，直到Spark程序运行完成后才释放退出。
      总体上看，Spark采用的是经典的scheduler/workers模式，每个Spark应用程序运行的第一步是构建一个可重用的资源池，然后在这个资源池里运行所有的ShuffleMap Task和Reduce Task（注意，尽管Spark引擎内部只用两类Task便可表示出一个复杂的应用程序，即ShuffleMap Task和Reduce Task），而MapReduce应用程序则不同，它不会构建一个可重用的资源池，而是让每个Task动态申请资源，且运行完后马上释放资源。

参看文章：
[1] http://www.raincent.com/content-85-11052-1.html
[2] http://www.raincent.com/content-85-11090-1.html
[3] http://www.raincent.com/content-85-11118-1.html
[4] https://www.cnblogs.com/rxingyue/p/7113100.html
[5] https://blog.csdn.net/andybegin/article/details/78190807