类似 MapReduce 的任务调度，Spark 也有自身的资源调度方式，包含应用程序间和应用程序中两个层面的策略调度。
每个 Spark 程序(SparkContext 实例)运行一个独立的执行进程，集群管理器提供了应用程序间的调度处理措施。其次，在每个 Spark 应用程序中，不同线程提交的多个 Job 作业(Spark Actions)可以同时运行。
作业调度
常见作业调度的基本概念：
- Job : 作业，一次 Action 生成一个 Job，一个 Job 是由一个或多个 Stage 组成的一次计算作业。
- Stage : 调度阶段，不存在 Shuffle 操作的调度过程，一个 Stage 各个算子之间都是窄依赖，一个 Stage 对应一个 TaskSet。
- TaskSet : 任务集，一组互相关联的，且没有 Shuffle 依赖关系的任务组成的集合。TaskSet 的大小和 Partition 大小有关。
- Task : 任务，单个分区数据集上的最小处理单元。根据 Stage 有多少个分区决定将 Stage 划分成多少个 Task，然后生成 TaskSet 放到 TaskScheduler。如果 Spark 为任务指定了 20 个 cores，Spark 就可以并发处理 20 个 Task。
一个 Job 就对应一个 DAG 图，划分 DAG 图时，首先将整个 Job 划分成一个 FinalStage，从后往前回溯，每遇到 Shuffle 操作就划分一个新的 Stage 出来。
应用程序间的调度
每个运行在集群上的 Spark 应用程序都能得到一个独立的，仅用于运行任务和存储数据的 JVM。如果用户需要共享集群资源，可以通过集群管理器配置不同的选项来分配集群资源。
静态资源分配
在集群中最简单最有效的方式就是静态资源分配，每个应用程序在整个生命周期中都可以得到一个最大数量的资源。此方式被用于Spark的Standalone和YARN模式中。
- Standalone模式资源分配：Spark应用程序按照FIFO顺序执行，每个应用程序都会尝试使用使用所有可用的节点。通过设置内核数目可以限制Spark应用程序资源的使用情况。
spark.deploy.defalutCores配置通用资源使用。单个应用程序特殊资源使用可以通过spark.cores.max修改配置。除了控制cpu cores之外，每个应用程序的spark.executor.memory配置可以控制每个Executor内存的使用。 - YARN模式的资源分配：在Spark YARN提交的客户端(spark-submit)配置参数。
--num-executors控制在集群上分配的Executors数量。--executor-memory控制Executor的内存分配。--executor-cores控制Executor的cpu分配。
动态资源分配
根据工作负载动态分配集群资源。在应用程序资源空闲时及时释放回集群，应用程序资源紧张时及时向集群申请空闲资源。这在集群中存在多个Spark应用程序同时运行时尤为重要。
使用动态资源分配时，必须要保存Executors在执行过程中写得shuffle文件，使Executors能够顺利释放，应用程序必须使用外部的shuffle服务。
- 配置外部shuffle服务的方式：在Spark YARN集群模式下，配置
spark.dynamicAllocation.enabled为true启用动态资源分配，并通过*.minExecutors和*.maxExecutors提供Executors的上下限。通过spark.shuffle.service.enabled为true启用外部shuffle服务，并在集群所有nodemanager上配置shuffle服务。 - 资源分配策略：Spark在资源空闲时释放Executors，在需要时重新获取Executors。由于没有一个明确的方法去预测一个将要释放资源的Executor是否将要运行一个新的任务，也没有办法预测当前的任务是否需要申请新增Executor，所以需要一组策略来决定什么时候释放或者申请Executors。
- 请求策略：当一个能动态分配资源的Spark应用程序有任务等待分配时，能够请求额外的Executors，这种情况意味着现有Executors不足以完成所有已经提交但是尚未完成的Task。Spark以轮询的方式申请Executors资源。当任务等待的时间超过
*.schedulerBacklogTimeout设定的时间，并且再次触发*.sustainedSchedulerBacklogTimeout延时之后，如果队列中处于等待的任务依然存在，申请额外Executors资源的请求就会被触发。而且每次轮询请求的Executors数目成倍增加。例如一个应用程序在第一次请求的时候会得到一个Executor，之后每轮会申请2，4，8，16……个Executors。这样做的动机有两个。一是一个应用程序在开始的时候应该谨慎的申请资源，避免一次申请过多造成资源浪费；二是应用程序能够及时的提供其资源使用情况，避免资源浪费。 - 删除策略：当Executor的空闲时间超过
*.executorIdleTimeout设定的值时，一个Spark应用程序将释放此Executor。删除策略和请求策略是互斥的，因为如果有等待的Executor资源的任务存在，就不会存在空闲的Executor。 - Executors优雅的退出机制：资源动态分配之前，当Executor执行失败，或Executor相关联应用程序退出时，一个Spark的Executor会退出，在这两种场景中，与Executor相关联的所有状态都不需要，可以安全的丢弃。但是，当一个Executor被显示删除时，应用程序仍然在运行。此时，应用程序试图访问Executor存储或写的状态，只能重新计算。为避免重新计算，需要在删除一个Executor之前保留状态。例如在shuffle期间，Spark的Executor会将自己的Map本地输出到磁盘。对于一些比较慢的事件，任务运行的时间比其他的时间更久，动态分配可能会在shuffle完成之前将空间的Executor删除掉，因为已经将状态保存在磁盘中，就不需要再重新计算了。保留shuffle文件需要使用外部Shuffle服务，这些服务是指长时间独立运行在集群每个节点的Spark应用程序以及Executors进程。启用该服务时，Spark的Executors将会从服务中获取shuffle文件以替代从备份中获取。这样任何一个Executor写得shuffle状态都可以超出Executor的生命周期继续服务。除了保留shuffle文件，Executor还可以将数据缓存在内存或者硬盘中，但是当Executor被删除后，这些缓存的数据就不能被访问了。
应用程序中的调度
在一个Spark应用程序中，如果在不同的线程中，多个并行的Job可以同时运行。因为Spark的任务调度是线程安全的，同时支持应用程序使用Case服务多个请求（多用户查询）。
Spark默认按照FIFO的方式运行Jobs，每个Job会被划分成多个Stages，分配可用资源的时候第一个Job的优先级最高，如果运行中的Job不需要使用所有的集群资源，后面的Job就不需要等待资源，可以立即执行。
Spark调度还支持以轮询的方式让不同的Job共享集群资源，此种方式称为公平调度。这种方式下，所有Job获得一个大约相当的集群资源。
启用公平调度器的方式：
val conf = new SparkConf()
conf.set("spark.scheduler.mode","FAIR")
公平调度池
Spark的公平调度器参考Hadoop的公平调度器，支持Job分组进入调度池，每个调度池可以设置不同的调度参数，同一分组内部按照用户均等分配资源，而不是按照Job平均共享资源。
新提交的任务通过下面的方式配置调度池
val sc = new SparkContext()
sc.setLocalProperty("spark.scheduler.pool","poolname")
如果不设置，任务会进入一个默认池。
设置调度池的属性之后，该线程中，被提交的全部Jobs(RDDs的save、collect、count等)都使用该Pool的名称。这样使得一个线程运行同一个用户的多个Jobs变得容易。
清除一个线程关联的池，调用
sc.setLocalProperty("spark.scheduler.pool",null)
配置公平调度池属性
默认情况下，每个池获得相同份额的集群资源，但是每个池中，Jobs以FIFO的方式运行。
池属性可以通过配置文件进行修改。
- schedulingMode，调度模式，FIFO或者FAIR
- weight，权重，相对与其他池的资源获取情况，默认所有池的权重都为1。如果一个池的权重设置为2，将获得权重为1的池的2倍资源。设置高权重的池中的Job总是会优先执行。
- minShare，最小共享。在按整体权重分配额外资源之前，公平调度器会为每个池分配的最小的共享资源（CPU核数）。默认每个池的minShare为0。
这些属性通过xml文件配置，默认文件类似于conf/fairscheduler.xml.template。要修改默认文件使用conf.set("spark.scheduler.allocation.file","path/fileName")。
XML文件格式如下：
<allocations>
<pool name="production">
<schedulingMode>FAIR</schedulingMode>
<weight>1</weight>
<minShare>2</minShare>
</pool>
<pool name="test">
<schedulingMode>FIFO</schedulingMode>
<weight>2</weight>
<minShare>3</minShare>
</pool>
</allocations>
调度器
DAGScheduler 负责构建具有依赖关系的任务集 TaskSet，TaskScheduler 负责将 TaskSet 资源提供给 TaskSetManager 供其作为调度任务的依据，TaskSetManager 负责在具体任务集的内部调度任务。
每一个 SparkContext 可能同时存在多个可运行的任务集（没有依赖关系），这些任务集之间的调度是由调度池 Pool 对象决定的，调度池 Pool 管理的对象是下一级的 Pool 或者 TaskSetManager 对象。
调度池
TaskSchedulerImpl在初始化时会根据用户设定的SchedulingMode（默认为FIFO）创建一个rootPool根调度池，之后根据具体的调度模式再进一步创建SchedulableBuilder对象，具体的SchedulableBuilder对象中的BuildPool方法将在rootPool的基础上完成整个Pool的构建工作。
调度池是在SparkContext内部的调度，多个Spark应用程序之间在调度池层面是没有调度优先级关系的。
两种调度模式，对应了两种类型的Pool
- FIFO：FIFO Pool直接管理TaskSetManager，每个TaskSetManager创建时都存储了其对应的StageID，FIFO Pool根据StageID的顺序调度TaskSetManager
- FAIR：FAIR Pool直接管理的对象是下一级的Pool，或者TaskSetManager，公平调度的基本原则是根据所管理的Pool/TaskSetManager中正在运行的任务的数量判断优先级，用户可以设置minShare最小任务数和weight任务权重调整对应Pool中任务集的优先程度。公平调度模式下构建的调度池是两级结构，根调度池管理一组子调度池，子调度池进一步管理属于该调度池的TaskSetManager。
