Spark?的资源管理架构

首先介绍一下?Spark?的资源管理架构。Spark?集群考虑到了未来对接一些更强大的资源管理系统（如?Yarn、Mesos?等），没有在资源管理的设计上对外封闭，所以Spark?架构设计时将资源管理抽象出了一层，通过这种抽象能够构建一种插件式的资源管理模块。

?

见上图，Master?是?Spark?的 主控节点，在实际的生产环境中会有多个?Master，只有一个?Master?处于?active?状态。Worker?是?Spark?的工作节点，向?Master?汇报自身的资源、Executeor?执行状态的改变，并接受?Master?的命令启动?Executor?或?Driver。Driver?是应用程序的驱动程序，每个应用包括许多小任务，Driver?负责推动这些小任务的有序执行。Executor?是?Spark?的工作进程，由?Worker 监管，负责具体任务的执行。

?

Spark 相关术语

master和worker节点

整个 Spark 集群中，分为 Master 节点与 worker 节点，同时一个集群有多个master节点和多个worker节点。

• master：主节点，该节点负责管理worker节点，我们从master节点提交应用，负责将串行任务变成可并行执行的任务集Tasks，同时还负责出错问题处理等；
• worker：从节点，该节点与master节点通信，负责执行任务并管理executor进程。它为集群中任何可以运行Application代码的节点，在Standalone模式中指的是通过slave文件配置的Worker节点，在Spark on Yarn模式下就是NoteManager节点

一台机器可以同时作为master和worker节点，比如有四台机器，可以选择一台设置为master节点，然后剩下三台设为worker节点，也可以把四台都设为worker节点，这种情况下，有一个机器既是master节点又是worker节点。

Application

Appliction都是指用户编写的Spark应用程序，其中包括一个Driver功能的代码和分布在集群中多个节点上运行的Executor代码

driver和executor进程

• Driver的功能是创建 SparkContext，负责执行用户写的 Application 的 main 函数进程，创建SparkContext的目的是为了准备Spark应用程序的运行环境，在Spark中有SparkContext负责与Cluster Manager通信，进行资源申请、任务的分配和监控等，当Executor部分运行完毕后，Driver同时负责将SparkContext关闭，通常用SparkContext代表Driver。不同的模式可能会将 Driver 调度到不同的节点上执行。
• executor：执行器，为某个Application运行在worker节点上的一个进程，该进程负责运行某些Task，并且负责将数据存到内存或磁盘上，每个Application都有各自独立的一批Executor进程。executor宿主在worker节点上，每个 Worker 上存在一个或多个 Executor 进程，每个executor持有一个线程池，每个线程可以执行一个task。根据 Executor 上 CPU-core 的数量，其每个时间可以并行多个跟 core 一样数量的 task。task 任务即为具体执行的 Spark 程序的任务。executor执行完task以后将结果返回给driver，每个executor执行的task都属于同一个应用。此外executor还有一个功能就是为应用程序中要求缓存的 RDD 提供内存式存储，RDD 是直接缓存在executor进程内的，因此任务可以在运行时充分利用缓存数据加速运算。

当我们在代码中执行了cache/persist等持久化操作时，根据我们选择的持久化级别的不同，每个task计算出来的数据也会保存到Executor进程的内存或者所在节点的磁盘文件中。

因此Executor的内存主要分为三块：第一块是让task执行我们自己编写的代码时使用，默认是占Executor总内存的20%；第二块是让task通过shuffle过程拉取了上一个stage的task的输出后，进行聚合等操作时使用，默认也是占Executor总内存的20%；第三块是让RDD持久化时使用，默认占Executor总内存的60%。

task的执行速度是跟每个Executor进程的CPU core数量有直接关系的。一个CPU core同一时间只能执行一个线程。而每个Executor进程上分配到的多个task，都是以每个task一条线程的方式，多线程并发运行的。如果CPU core数量比较充足，而且分配到的task数量比较合理，那么通常来说，可以比较快速和高效地执行完这些task线程。

Cluter Manager

集群管理器，指的是在集群上获取资源的外部服务。目前有三种类型：

1. Standalone : spark原生的资源管理，由Master负责资源的分配，易于构建集群
2. Apache Mesos：通用的集群管理，与hadoop MR兼容性良好的一种资源调度框架，可以在其上运行Hadoop MapReduce和一些服务应用
3. Hadoop Yarn: 主要是指Yarn中的ResourceManager

在集群不是特别大，并且没有mapReduce和Spark同时运行的需求的情况下，用Standalone模式效率最高。

Task

被送到某个Executor上的工作单元，是运行Application的基本单位，多个Task组成一个Stage，而Task的调度和管理等是由TaskScheduler负责

Job

包含多个Task组成的并行计算，往往由Spark Action触发生成， 一个Application中往往会产生多个Job。总之Job=多个stage

Stage

每个Job会被拆分成多组Task， 作为一个TaskSet， 其名称为Stage，Stage的划分和调度是有DAGScheduler来负责的，Stage有非最终的Stage（Shuffle Map Stage）和最终的Stage（Result Stage）两种，Stage的边界就是发生shuffle的地方。总之Stage=多个同种task

DAGScheduler

根据Job构建基于Stage的DAG（Directed Acyclic Graph有向无环图)，并提交Stage给TASkScheduler。 其划分Stage的依据是RDD之间的依赖的关系找出开销最小的调度方法

TASKSedulter

将TaskSET提交给worker运行，每个Executor运行什么Task就是在此处分配的。TaskScheduler维护所有TaskSet，当Executor向Driver发生心跳时，TaskScheduler会根据资源剩余情况分配相应的Task。另外TaskScheduler还维护着所有Task的运行标签，重试失败的Task。下图展示了TaskScheduler的作用：

?

在不同运行模式中任务调度器具体为：

1. Spark on Standalone模式为TaskScheduler
2. YARN-Client模式为YarnClientClusterScheduler
3. YARN-Cluster模式为YarnClusterScheduler

Spark作业基本运行原理

第一步：当我们提交一个Spark作业之后，这个作业就会启动一个对应的Driver进程。driver进程就是应用的main()函数并且构建sparkContext对象，根据使用的部署模式不同，Driver进程可能在本地启动，也可能在集群中某个工作节点上启动。driver本身会根据我们设置的参数占有一定的资源（主要指cpu core和memory）。

第二步：Driver进程首先会向集群管理器（standalone、yarn，mesos）申请spark应用所需的资源，这里的资源指的就是Executor进程。然后集群管理器会根据spark应用所设置的参数在各个worker上分配一定数量的executor，每个executor都占用一定数量的cpu和memory。

第三步：在得到申请的应用所需资源以后，driver就开始调度和执行我们编写的应用代码。driver进程会将我们编写的spark应用代码拆分成多个stage，每个stage执行一部分代码片段，并为每个stage创建一批task，然后将这些tasks分配到各个executor中执行，task是最小的计算单元，负责执行一模一样的计算逻辑（也就是我们自己编写的某个代码片段），只是每个task处理的数据不同而已。

第四步：一个stage的所有task都执行完毕之后，会在各个节点本地的磁盘文件中写入计算中间结果，然后Driver就会调度运行下一个stage。下一个stage的task的输入数据就是上一个stage输出的中间结果。如此循环往复，直到将我们自己编写的代码逻辑全部执行完，并且计算完所有的数据，得到我们想要的结果为止。运行完成后，会释放所有资源。

Spark运行特点

1. 每个Application获取专属的executor进程，该进程在Application期间一直驻留，并以多线程方式运行Task。这种Application隔离机制是有优势的，无论是从调度角度看（每个Driver调度他自己的任务），还是从运行角度看（来自不同Application的Task运行在不同JVM中），当然这样意味着Spark Application不能跨应用程序共享数据，除非将数据写入外部存储系统
2. Spark与资源管理器无关，只要能够获取executor进程，并能保持相互通信就可以了
3. 提交SparkContext的Client应该靠近Worker节点（运行Executor的节点），最好是在同一个Rack里，因为Spark Application运行过程中SparkContext和Executor之间有大量的信息交换
4. Task采用了数据本地性和推测执行的优化机制
5. 使用多线程池模型减少 Task 启动开稍, shuffle 过程中避免不必要的 sort 操作并减少磁盘 IO 操作。(Hadoop 的 Map 和 reduce 之间的 shuffle 需要 sort)
6. 提供 Cache 机制来支持需要反复迭代计算或者多次数据共享，减少数据读取的 IO 开销
7. 提供了一套支持 DAG 图的分布式并行计算的编程框架，减少多次计算之间中间结果写到 Hdfs 的开销；
8. Spark是根据shuffle类算子来进行stage的划分。如果我们的代码中执行了某个shuffle类算子（比如reduceByKey、join等），那么就会在该算子处，划分出一个stage界限来。可以大致理解为，shuffle算子执行之前的代码会被划分为一个stage，shuffle算子执行以及之后的代码会被划分为下一个stage。因此一个stage刚开始执行的时候，它的每个task可能都会从上一个stage的task所在的节点，去通过网络传输拉取需要自己处理的所有key，然后对拉取到的所有相同的key使用我们自己编写的算子函数执行聚合操作（比如reduceByKey()算子接收的函数）。这个过程就是shuffle

参考：

Spark基本架构及原理

Spark on Yarn ，了解一下？???????

本文地址：https://blog.csdn.net/yawei_liu1688/article/details/111958747