Java大数据开发入门系列(四)————Spark之作业调度

简介

Spark 有几种计算资源调度的方式。前面讲到，每个Spark应用（SparkContext的实例）都运行一组独立的executor进程。spark运行的集群管理器提供了跨应用之间的资源调度scheduling across applications。其次，在每个Spark应用中，如果多个 jobs（Spark算子）是由不同的线程提交的，那么它们可能会并发运行。如果你的应用是常见的通过网络服务请求，Spark包含一个fair scheduler(公平的调度器)来调度每个SparkContext内的资源。在 Spark 应用内部（对应同一个 SparkContext）各个作业之间，Spark 默认 FIFO 调度，同时也可以支持公平调度。

跨应用调度

当在集群上运行时，每个Spark应用都会得到一组独立的executorJVM，来运行其任务并存储数据。如果多个用户需要共享你的集群，那么会有很多资源分配相关的选项，如何设计还取觉于具体的集群管理器。

对 Spark 所支持的各个集群管理器而言，最简单的的资源分配，就是静态划分。这种方式就意味着，每个 Spark 应用都是设定一个最大可用资源总量，并且该应用在整个生命周期内都会占住这个资源。这种方式在 Spark独立部署 standalone 和 YARN调度，以及 Mesos 粗粒度模式(coarse-grained Mesos mode)下都可用。资源分配可以根据集群类型进行如下配置：

• Standalone mode: 默认情况下，提交到standalone mode集群的应用将以FIFO（先进先出）的顺序运行，并且每个spark应用都会占用集群中所有可用节点。不过你可以通过设置 spark.cores.max 或者 spark.deploy.defaultCores 来限制单个应用所占用的节点个数。最后，除了可以控制对 CPU 的使用数量之外，还可以通过spark.executor.memory来控制各个应用的内存占用量。
• Mesos: 在Mesos中要使用静态划分的话，需要将 spark.mesos.coarse 设为true，同样，你也需要配置 spark.cores.max来控制各个应用的 CPU 总数，以及 spark.executor.memory 来控制各个应用的内存占用。
• YARN: 在 YARN 中需要使用 –num-executors 选项来控制 Spark 应用在集群中分配的执行器的个数。对于单个执行器（executor）所占用的资源，可以使用 –executor-memory 和 –executor-cores 来控制。

Mesos 上还有一种动态共享 CPU 的方式。在这种模式下，每个 Spark 应用的内存占用仍然是固定且独占的(仍由 spark.exexcutor.memory 决定)，但是如果该 Spark 应用没有在某个机器上执行任务的话，那么其它应用可以占用该机器上的 CPU。这种模式对集群中有大量不是很活跃应用的场景非常有效，例如：集群中有很多不同用户的 Spark shell session。但这种模式不适用于低延时的场景，因为当 Spark 应用需要使用 CPU 的时候，可能需要等待一段时间才能取得对 CPU 的使用权。要使用这种模式，只需要在 mesos://URL 上设置 spark.mesos.coarse 属性为 false 即可。

值得注意的是，目前还没有任何一种资源分配模式支持跨 Spark 应用的内存共享。如果你想通过这种方式共享数据，我们建议你可以单独使用一个服务(例如：alluxio)，这样就能实现多应用访问同一个 RDD 的数据。

动态资源分配

Spark 提供了一种基于负载来动态调节 Spark 应用资源占用的机制。这意味着，你的应用会在资源空闲的时间将其释放给集群，需要时再重新申请。这一特性在多个应用 Spark 集群资源的情况下特别有用。

这个特性默认是禁止的，但是在所有的粗粒度集群管理器上都是可用的，如：独立部署模式standalone mode，YARN mode，和Mesos 粗粒度模式(coarse-grained Mesos mode)。

配置和部署

要使用动态资源分配这一特性有两个前提条件。首先，你的应用必须设置 spark.dynamicAllocation.enabled 为 true。其次，你必须在每个worker节点上启动 external shuffle service，同时将 spark.shuffle.service.enabled 设为 true。external shuffle service 的目的是在移除 executor 的时候，能够保留 executor 输出的 shuffle 文件。启用 external shuffle service 的方式在各个集群管理器上各不相同：

在 Spark 独立部署的集群中，你只需要在 worker 启动前设置 spark.shuffle.service.enabled 为 true 即可。

在 Mesos 粗粒度模式下，你需要在各个节点上运行 $SPARK_HOME/sbin/start-mesos-shuffle-service.sh 并设置 spark.shuffle.service.enabled为true即可。例如，你可以在Marathon来启用这一功能。

在YARN模式下，需要按以下步骤在各个 NodeManager 上启动：here

资源分配策略

总体上来说，Spark 应该在executor空闲时将其关闭，而在后续要用时再申请。因为没有一个固定的方法，可以预测一个executor在后续是否马上会被分配去执行任务，或者一个新分配的executor实际上是空闲的，所以我们需要一个试探性的方法，来决定是否申请或是移除一个executor。

请求策略

一个启用了动态分配的 Spark 应用在等待任务需要调度的时候，会去申请额外的executor。在这种情况下，必定意味着已有的executor已经不足以同时执行所有未完成的任务。

Spark会轮流来申请executor。当有待处理的任务达到spark.dynamicAllocation.scheduleerBacklogTimeout秒时，就会触发实际的请求，如果等待队列中仍有挂起的任务，则每过 spark.dynamicAllocation.sustainedSchedulerBacklogTimeout 秒后触发一次资源申请。另外，每一轮申请的executor个数以指数形式增长。例如：一个 Spark 应用可能在首轮申请 1 个执行器，后续的轮次申请个数可能是 2 个、4 个、8 个……。

采用指数级增长策略的原因有两个：第一，一个应用程序在开始时应该谨慎地请求executor，以防万一发现只增加几个executor就够了，这和 TCP 慢启动有些类似；第二，如果一旦 Spark 应用确实需要申请多个executor的话，那么可以确保其所需的计算资源及时增长。

移除策略

移除executor的策略就简单得多了。Spark 应用会在某个执行器空闲超过 spark.dynamicAllocation.executorIdleTimeout 秒后将其删除，在大多数情况下，执行器的移除条件和申请条件都是互斥的，也就是说，执行器在有等待执行任务挂起时，不应该空闲。

优雅的关闭Executor

在非动态分配下，executor可能的退出原因有执行失败或是相关 Spark 应用已经退出。不管是哪种原因，executor的所有相关联的状态都已经不再需要，可以丢弃掉。但是在动态分配的情况下，executor有可能在 Spark 应用运行期间被移除。这时候，如果 Spark 应用尝试去访问该executor存储的状态，就必须重算这一部分数据。因此，Spark 需要一种机制，能够优雅的关闭executor，同时还保留其状态数据。

这种需求对于shuffles操作尤其重要。shuffle过程中，Spark executor首先将 map 输出写到本地磁盘，然后在其他executor试图获取这些map结果数据时，充当这些文件的服务器。一旦有某些任务执行时间过长，动态分配有可能在shuffle结束前移除任务异常的executor，而这些被移除的executor对应的数据将会被重新计算，但这些重算其实是不必要的。

要解决这一问题，就需要用到 external shuffle service，该服务在 Spark 1.2 引入。该服务在每个节点上都会启动一个不依赖于任何 Spark 应用或executor的独立进程。一旦该服务启用，Spark executor不再从各个executor上获取 shuffle文件，转而从这个 service 获取。这意味着，任何执行器输出的混洗状态数据都可能存留时间比对应的执行器进程还长。

除了shuffle文件之外，executor也会在磁盘或者内存中缓存数据。一旦executor被移除，其缓存数据将无法访问。这个问题目前还没有解决。或许在未来的版本中，可能会类似采用external shuffle service的方法，将缓存数据保存在堆外存储中以解决这一问题。

应用内调度

在指定的 Spark 应用内部（对应同一 SparkContext 实例），多个线程可能并发地提交 Spark 作业（job），他们可以同时运行。在本节中，作业（job）是指，由 Spark action 算子（如：collect）触发的一系列计算任务的集合。Spark 调度器是完全线程安全的，并且能够支持 Spark 应用同时处理多个请求（比如：来自不同用户的查询）。

默认，Spark 应用内部使用 FIFO 调度策略。每个作业被划分为多个阶段（stage）（例如：map 阶段和 reduce 阶段），第一个作业在其启动后会优先获取所有的可用资源，然后是第二个作业再申请，再第三个……。如果前面的作业没有把集群资源占满，则后续的作业可以立即启动运行，否则，后提交的作业会有明显的延迟等待。

不过从 Spark 0.8 开始，Spark 也能支持各个作业间的公平（Fair）调度。公平调度时，Spark 以轮询的方式给每个作业分配资源，因此所有的作业获得的资源大体上是平均分配。这意味着，即使有大作业在运行，小的作业再提交也能立即获得计算资源而不是等待前面的作业结束，大大减少了延迟时间。这种模式特别适合于多用户配置。要启用公平调度器，只需设置一下 SparkContext 中 spark.scheduler.mode 属性为 FAIR 即可 :

val conf = new SparkConf().setMaster(...).setAppName(...)
conf.set("spark.scheduler.mode", "FAIR")
val sc = new SparkContext(conf)

公平调度资源池

公平调度器还可以支持将作业分组放入资源池（pool），然后给每个资源池配置不同的调度选项（如：权重）(ps:就是设置调度优先级)。这样你就可以给一些比较重要的作业创建一个“高优先级”资源池，或者你也可以把每个用户的作业分到一组，这样一来就是各个用户平均分享集群资源，而不是各个作业平分集群资源。Spark 公平调度的实现方式基本都是模仿 Hadoop Fair Scheduler来实现的。

默认情况下，新提交的作业都会进入到默认资源池中，不过作业对应于哪个资源池，可以在提交作业的线程中用 SparkContext.setLocalProperty 设定 spark.scheduler.pool 属性。示例代码如下 :

// Assuming sc is your SparkContext variable
sc.setLocalProperty("spark.scheduler.pool", "pool1")

一旦设好了局部属性，所有该线程所提交的作业（即：在该线程中调用 action 算子，如：RDD.save/count/collect 等）都会使用这个资源池。这个设置是以线程为单位保存的，你很容易实现用同一线程来提交同一用户的所有作业到同一个资源池中。同样，如果需要清除资源池设置，只需在对应线程中调用如下代码 :

sc.setLocalProperty("spark.scheduler.pool", null)

资源池默认行为

默认地，各个资源池之间平分整个集群的资源（包括 default 资源池），但在资源池内部，默认情况下，作业是 FIFO 顺序执行的。举例来说，如果你为每个用户创建了一个资源池，那么久意味着各个用户之间共享整个集群的资源，但每个用户自己提交的作业是按顺序执行的，而不会出现后提交的作业抢占前面作业的资源。

配置资源池属性

资源池的属性需要通过配置文件来指定。每个资源池都支持以下3个属性 :

• schedulingMode：可以是 FIFO 或 FAIR，控制资源池内部的作业是如何调度的。
• weight：控制资源池相对其他资源池，可以分配到资源的比例。默认所有资源池的 weight 都是 1。如果你将某个资源池的 weight 设为 2，那么该资源池中的资源将是其他池子的2倍。如果将 weight 设得很高，如 1000，可以实现资源池之间的调度优先级 – 也就是说，weight=1000 的资源池总能立即启动其对应的作业。
• minShare：除了整体 weight 之外，每个资源池还能指定一个最小资源分配值（CPU 个数），管理员可能会需要这个设置。公平调度器总是会尝试优先满足所有活跃（active）资源池的最小资源分配值，然后再根据各个池子的 weight 来分配剩下的资源。因此，minShare 属性能够确保每个资源池都能至少获得一定量的集群资源。minShare 的默认值是 0。

资源池属性是一个 XML 文件，可以基于 conf/fairscheduler.xml.template 修改，然后在 SparkConf。的 spark.scheduler.allocation.file 属性指定文件路径：

conf.set("spark.scheduler.allocation.file", "/path/to/file")

资源池 XML 配置文件格式如下，其中每个池子对应一个 元素，每个资源池可以有其独立的配置 :

<?xml version="1.0"?>
<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>

额外补充：

Shuffle相关介绍

概述

有些运算需要将各节点上的同一类数据汇集到某一节点进行计算，把这些分布在不同节点的数据按照一定的规则汇集到一起的过程称为 Shuffle。

在MapReduce框架，Shuffle是连接Map和Reduce之间的桥梁，Map阶段通过shuffle读取数据并输出到对应的Reduce，而Reduce阶段负责从Map端拉取数据并进行计算。在整个shuffle过程中，往往伴随着大量的磁盘和网络I/O。所以shuffle性能的高低也直接决定了整个程序的性能高低。而Spark也会有自己的shuffle实现过程。

Spark中的 shuffle

在DAG调度的过程中，Stage 阶段的划分是根据是否有shuffle过程，也就是存在 宽依赖 的时候,需要进行shuffle,这时候会将 job 划分成多个Stage，每一个 Stage 内部有很多可以并行运行的 Task。

标注对应的RDD后：


stage与stage之间的过程就是 shuffle 阶段，前一个 Stage 的 ShuffleMapTask 进行 Shuffle Write， 把数据存储在 BlockManager 上面， 并且把数据位置元信息上报到 Driver 的 MapOutTrack 组件中， 下一个 Stage 根据数据位置元信息， 进行 Shuffle Read， 拉取上个 Stage 的输出数据。在 Spark 中，负责 shuffle 过程的执行、计算和处理的组件主要就是 ShuffleManager 。ShuffleManager 随着Spark的发展有两种实现的方式，分别为 HashShuffleManager 和 SortShuffleManager ，因此spark的Shuffle有 Hash Shuffle 和 Sort Shuffle 两种。

在 Spark 1.2 以前，默认的shuffle计算引擎是 HashShuffleManager 。

HashShuffleManager 有着一个非常严重的弊端，就是会产生大量的中间磁盘文件，进而由大量的磁盘IO操作影响了性能。因此在Spark 1.2以后的版本中，默认的 ShuffleManager 改成了 SortShuffleManager 。

SortShuffleManager 相较于 HashShuffleManager 来说，有了一定的改进。主要就在于每个Task在进行shuffle操作时，虽然也会产生较多的临时磁盘文件，但是最后会将所有的临时文件合并(merge)成一个磁盘文件，因此每个 Task 就只有一个磁盘文件。在下一个 Stage 的shuffle read task拉取自己的数据时，只要根据索引读取每个磁盘文件中的部分数据即可。

Hash shuffle是不具有排序的Shuffle。

YARN启动shuffle服务

要在YARN群集中的每个NodeManager上启动Spark Shuffle服务，请遵循以下说明：

1. 用YARN配置文件构建Spark。如果你使用的是预打包的发行版，请跳过这一步。
2. 找到spark-<version>-yarn-shuffle.jar. 如果你是自己构建的spark会在 $SPARK_HOME/common/network-yarn/target/scala-<version> 下,如果你用的我是发行版则在 yarn 下.
3. 将这个jar包添加到集群中所有NodeManagers的classpath中。
4. 在每个节点的yarn-site.xml中，将spark_shuffle添加到yarn.nodemanager.aux-services中，然后将yarn.nodemanager.aux-services.spark_shuffle.class设置org.apache.spark.network.yarn.YarnShuffleService。
5. 在 etc/hadoop/yarn-env.sh 中设置 YARN_HEAPSIZE（默认为 1000），增加 NodeManager 的堆大小，以避免在shuffle中的垃圾回收问题。
6. 重新启动集群中的所有NodeManagers。

在YARN上运行shuffle service时，可以使用以下附加配置选项：

Property Name	Default	Meaning
spark.yarn.shuffle.stopOnFailure	false	当Spark shuffle service的初始化出现故障时，是否要停止NodeManager。这可以防止在Spark shuffle service没有运行的NodeManagers上运行容器导致的应用失败。

Spark调度算法(FIFO、FAIR)

FIFO模式的算法类是FIFOSchedulingAlgorithm，FAIR模式的算法实现类是FairSchedulingAlgorithm。下面看两种模式下的比较函数的实现，FIFO：

先比较priority，在FIFO中该优先级实际上是Job ID，越早提交的job的jobId越小，priority越小，优先级越高。

若priority相同，则说明是同一个job里的TaskSetMagager，则比较StageId，StageId越小优先级越高。

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
      res = math.signum(stageId1 - stageId2)
      //优先级相同，就比较stageId，StageId越小优先级越高
    }
    res < 0
  }

FAIR 模式中有一个 rootPool 和多个子 Pool， 各个子 Pool 中存储着所有待分配的 TaskSetMagager 。

在 FAIR 模 式 中 ， 需 要 先 对 子 Pool 进 行 排 序 ， 再 对 子 Pool 里 面 的

TaskSetMagager进行排序，因为Pool和TaskSetMagager都继承了Schedulable特质， 因此使用相同的排序算法。

排序过程的比较是基于 Fair-share 来比较的，每个要排序的对象包含三个属性:

runningTasks 值（ 正在运行的 Task 数）、minShare 值、weight 值，比较时会综合考量 runningTasks 值， minShare 值以及 weight 值。

注意，minShare、weight 的值均在公平调度配置文件 fairscheduler.xml 中被指定，调度池在构建阶段会读取此文件的相关配置。

1) 如果 A 对象的 runningTasks 大于它的 minShare， B 对象的 runningTasks 小于它的 minShare，那么 B 排在 A 前面； （ 如果一个Pool的miniShare够用，另一个不够用，先分配给够用的。）

2) 如果 A 、B 对象的 runningTasks 都小于它们的 minShare ，那么就比较runningTasks 与 minShare 的比值（ minShare 使用率），谁小谁排前面；（ 两个poll都够用，谁占miniShare的少先分配给谁。例如两个Pool同样数量的runningTask，先分配给miniShare大的。）

3) 如果 A 、B 对象的 runningTasks 都大于它们的 minShare ，那么就比较runningTasks 与 weight 的比值（ 权重使用率），谁小谁排前面。（同样数量的runningTask，先分配给weight大的）

4) 如果上述比较均相等，则比较名字。

整体上来说就是通过 minShare 和 weight 这两个参数控制比较过程， 可以做到让 minShare 使用率和权重使用率少（ 实际运行 task 比例较少） 的先运行。

private[spark] class FairSchedulingAlgorithm extends SchedulingAlgorithm {
  override def comparator(s1: Schedulable, s2: Schedulable): Boolean = {
    val minShare1 = s1.minShare
    val minShare2 = s2.minShare
    //默认为0,除非通过fair的配置文件进行了配置指定
 
    val runningTasks1 = s1.runningTasks
    val runningTasks2 = s2.runningTasks
    /* 如果是TaskSetManager时,就是taskSet中运行的task的个数,
     * 如果是Pool实例是表示是所有使用这个poolName的所有的TaskSetManager正在运行的task的个数.
     */
 
    val s1Needy = runningTasks1 < minShare1
    val s2Needy = runningTasks2 < minShare2
    //只有在minShare在fair的配置文件中显示配置,同时大于正在运行的task的个数时,才会为true
 
    val minShareRatio1 = runningTasks1.toDouble / math.max(minShare1, 1.0).toDouble
    val minShareRatio2 = runningTasks2.toDouble / math.max(minShare2, 1.0).toDouble
    //运行的task的个数针对于minShare的比重
    
    val taskToWeightRatio1 = runningTasks1.toDouble / s1.weight.toDouble
    val taskToWeightRatio2 = runningTasks2.toDouble / s2.weight.toDouble
    //得到正在运行的task个数针对于pool的weight的比重
    var compare: Int = 0
 
    /*这里首先根据正在运行的task的个数是否已经达到调度队列中最小的分片的个数来进行排序,
    * 如果s1中运行运行的个数小于s1的pool的配置的minShare,返回true,表示s1排序在前面.
    * 如果s2中运行的task的个数小于s2的pool中配置的minShare(最小分片数)的值,表示s1小于s2,这时s2排序应该靠		 * 前.
    */
 
    if (s1Needy && !s2Needy) {
      return true
    } else if (!s1Needy && s2Needy) {
      return false
    } else if (s1Needy && s2Needy) {
    /*这种情况表示s1与s2两个队列中,正在运行的task的个数都已经大于(不小于)了两个子调度器中配置的minShare的   		* 个数时,根据两个子调度器队列中正在运行的task的个数对应此调度器中最小分片的值所占的比重最小的一个排序更靠前
    */
      compare = minShareRatio1.compareTo(minShareRatio2)
    } else {
    /*这种情况表示s1与s2两个子调度器的队列中,正在运行的task的个数都还没有达到配置的最小分片的个数的情况,比
    *	较两个队列中正在运行的task的个数对应调度器队列的weigth的占比,最小的一个排序更靠前
    */
      compare = taskToWeightRatio1.compareTo(taskToWeightRatio2)
    }
 
    if (compare < 0) {
      true
    } else if (compare > 0) {
      false
    } else {
    //如果两个根据上面的计算,排序值都相同,就看看这两个调度器的名称,按名称的字节序来排序了.
      s1.name < s2.name
    }
  }
}

参考文献：

• Spark的Shuffle总结分析-说出你的愿望吧
• Spark Shuffle 详解-小鹏
• Job-scheduling-apache
• SPARK的TASK调度器(FAIR公平调度算法)-herman很慢