版本：2.3.0

Overview

Spark有几个工具用于在计算之间调度资源。首先回想一下，如集群模式概述中所述，每个Spark应用程序（SparkContext的实例）运行一组独立的executor进程。Spark运行的集群管理器提供跨应用程序调度。其次， 在每个Spark应用程序中，如果多个“作业”（Spark actions）由不同的线程提交，它们可能会同时运行。这对于通过网络提供服务应用程序来说很常见。Spark包含一个公平的调度器来调度每个SparkContext中的资源。

在应用程序间调度

在群集上运行时，每个Spark应用程序都会获得一组独立的executor JVM，它们仅为该应用程序运行task和存储数据。如果多个用户需要共享群集，则根据群集管理器的不同，可以使用不同的选项来管理分配。

对所有集群管理器都可以使用的最简单的选项是资源的静态分区。采用这种方法，每个应用程序都可以使用最大量的资源，并在整个持续时间内保留它们。这是Spark 独立 和YARN模式以及 粗粒度Mesos模式中使用的方法。资源分配可以根据集群管理器的类型进行如下配置：

• 独立模式：默认情况下，提交给独立模式集群的应用程序将以FIFO（先进先出）顺序运行，并且每个应用程序将尝试使用所有的可用节点。您可以通过设置spark.cores.max属性来限制一个应用程序使用的节点数量，或设置应用程序的默认值spark.deploy.defaultCores。最后，除了控制内核之外，每个应用程序的spark.executor.memory还控制着内存的使用。
• Mesos：要在Mesos上使用静态分区，请将spark.mesos.coarse属性设置为true，并且可以选择设置spark.cores.max参数，类似独立模式。您还应该设置spark.executor.memory为控制执行程序的内存。
• YARN： Spark YARN客户端使用--num-executors参数选项控制集群上分配的执行者数量（同spark.executor.instances，该参数是作为配置属性配置在文件中的），而--executor-memory （spark.executor.memory配置属性）和--executor-cores（spark.executor.cores配置属性）控制每个执行者的资源。有关更多信息，请参阅 YARN Spark属性。

Mesos上的第二个选项是动态共享 CPU内核。在这种模式下，每个Spark应用程序仍然具有固定且独立的内存分配（由其设置spark.executor.memory），但是当应用程序不在计算机上运行任务时，其他应用程序可能会在这些内核上运行任务。当您期望大量不是过度活跃的应用程序时（例如来自不同用户的shell会话），此模式非常有用。但是，它具有较低的可预测延迟风险，因为应用程序在需要时可能需要一段时间才能获得一个节点上的内核。要使用此模式，只需使用 mesos://URL并设置spark.mesos.coarse为false。

请注意，目前没有任何模式提供跨应用程序的内存共享。如果您希望以这种方式共享数据，我们建议运行一个服务器应用程序，通过查询相同的RDD来服务多个请求。（是不是类似spark 整合 redis ？）

动态资源分配

Spark提供了一种机制，可根据工作负载动态调整应用程序占用的资源。这意味着如果您的应用程序不再使用时可能会将资源返回给群集，并且在有需求时再次请求它们。如果多个应用程序共享Spark群集中的资源，此功能特别有用。

此功能在默认情况下处于禁用状态，并在所有粗粒度集群管理器（即独立模式，YARN模式和 Mesos粗粒度模式）上可用 。

配置和设置

使用此功能有两个要求。首先，你的应用程序必须设置 spark.dynamicAllocation.enabled为true。其次，您必须 在同一集群中的每个工作节点上设置external shuffle service，并设置spark.shuffle.service.enabled为true。external shuffle service 的目的是允许删除执行者，而不删除由他们编写的shuffle文件（更详细描述 如下）。设置此服务的方式因集群管理器而异：

• 在独立模式下，只需配置spark.shuffle.service.enabled为true。

• 在Mesos粗粒度模式下，在所有的节点slave上运行$SPARK_HOME/sbin/start-mesos-shuffle-service.sh，并设置spark.shuffle.service.enabled 为true。

• 在YARN模式下，请按照此处的说明进行操作。
  （即：配置yarn-site.xml 的 的spark_shuffle ）

位于spark.dynamicAllocation.*和 spark.shuffle.service.*命名空间下所有其他相关配置都是可选的。有关更多详细信息，请参阅 配置页面。

资源分配策略

在很高的层面上，Spark应该在不再使用时放弃执行者，并在需要时收购执行者。由于没有确切的方法可以预测即将被移除的执行者是否会在不久的将来执行任务，或者即将添加的新执行者是否实际上是空闲的，我们需要一组启发式来确定何时删除或请求执行者。

请求政策

启用动态分配的Spark应用程序，在有任务等待被调度时请求执行者。这种情况必然意味着现有的执行者不足饱和执行已提交但尚未完成的任务。

Spark循环请求执行者。当挂起的任务持续spark.dynamicAllocation.schedulerBacklogTimeout数秒时会触发请求，若任务队列持续存在，则每spark.dynamicAllocation.sustainedSchedulerBacklogTimeout秒钟后再次触发。另外，每轮请求的执行者数量比前一轮增加指数级。例如，一个应用程序会在第一轮添加1个执行程序，然后在后续的循环中添加2，4，8等执行程序。

指数式增长政策的动机是双重的。首先，应用程序应该在开始时谨慎地请求执行者，以防只有少数额外的执行者就足够了。这反映了TCP慢启动的理由。其次，应用程序应该能够及时提高资源使用量，以防万一实际需要许多执行者。

删除政策

去除执行者的策略要简单得多。Spark应用程序在执行器闲置超过spark.dynamicAllocation.executorIdleTimeout秒钟时将其删除。请注意，在大多数情况下，这种情况与请求条件是互斥的，因为如果仍有待处理的任务被安排，则执行者不应闲置。

执行者的退役

在动态分配之前，Spark执行者会在失败或关联应用程序退出时退出。在这两种情况下，与执行器关联的所有状态不再需要，可以安全地丢弃。但是，使用动态分配，当执行者被明确删除时，应用程序仍在运行。如果应用程序试图访问执行程序中存储或写入的状态，则必须重新计算状态。因此，Spark需要一种机制，通过在删除执行者之前保留其状态，优雅地停用执行者。

这个要求对shuffle特别重要。在shuffle期间，Spark执行者首先将自己的映射输出写入磁盘，然后在其他执行者试图获取它们时充当这些文件的服务器。如果执行者的任务运行时间比同伴要长得多，动态分配可能会在shuffle完成之前删除执行者，在这种情况下，该执行者写入的shuffle文件必须重新计算。

保留shuffle文件的解决方案是使用Spark 1.2中引入的外部shuffle服务。此服务是指在群集中的每个节点上运行的独立于Spark应用程序及其执行程序的长时间运行的进程。如果服务已启用，则Spark执行程序将从服务中而不是从其他服务中获取shuffle文件。这意味着由执行者编写的任何混乱状态可以继续在执行者之外服务。

除了写入shuffle文件之外，执行者还可以将数据缓存在磁盘或内存中。但是，当执行者被删除时，所有缓存的数据将不再可用。为了缓解这种情况，默认情况下，不会删除包含缓存数据的执行者。您可以使用配置此行为 spark.dynamicAllocation.cachedExecutorIdleTimeout。在将来的版本中，缓存数据可以通过离堆存储来保存，这与通过外部shuffle服务如何保留随机文件的精神类似。

在应用程序中调度

在一个给定的Spark应用程序（SparkContext实例）中，如果多个并行作业是从不同的线程提交的，它们可以同时运行。这里的“job作业”，指的是spark的动作（例如save， collect）和任何需要运行以返回的action。Spark的调度程序完全是线程安全的，并且支持服务多个请求（例如，针对多个用户的查询）。

默认情况下，Spark的调度程序以FIFO方式运行作业。每个工作被分为“阶段”（例如map和reduce阶段），第一份作业优先使用所有可用资源，然后第二份工作优先等等。如果队列首部的作业不需要使用整个集群，后面的作业可以立即开始运行，但是如果队列头部的作业很大，则后面的作业可能会显着延迟。

从Spark 0.8开始，也可以在作业之间配置公平共享。在公平分享下，Spark以“循环”方式在作业之间分配任务，以便所有作业获得大致相等的群集资源份额。这意味着在长工作运行的同时提交的短工可以立即接收资源，可以获得良好的响应时间，而无需等待长工作完成。该模式最适合多用户设置。

要启用公平调度程序，只需在配置SparkContext时将该spark.scheduler.mode属性设置为FAIR：

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

公平调度池

公平调度程序还支持将作业分组到pool，并为每个池设置不同的调度选项（例如权重weight）。例如，为更重要的工作创建“高优先级”池，或者将每个用户的工作组合在一起，并为用户提供相同的份额，而不管他们有多少并发作业，而不是提供作业平等份额。这种方法是在Hadoop Fair Scheduler之后建模的 。

在没有任何干预的情况下，新提交的作业会进入default pool，但可以通过为Sparkcontext添加spark.scheduler.pool“pool属性” 来设置作业池。这如下完成：

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

设置这个pool属性后，所有的工作（在此线程通过调用这个线程内提交RDD.save，count，collect，等）将使用此池名称。该设置是按线程进行的，以便让代表同一用户的线程可以轻松运行多个作业。如果您想清除线程关联的池，只需调用：

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

池的默认行为

默认情况下，每个池获得集群的平等份额（也与默认池中的每个作业的份额相同），但在每个池中，作业按FIFO顺序运行。例如，如果您为每个用户创建一个池，这意味着每个用户将获得相同的群集份额，并且每个用户的查询将按顺序运行，而不是稍后的查询从该用户的较早的查询获取资源。

配置池属性

特定池的属性也可以通过配置文件进行修改。每个池支持三个属性：

• schedulingMode：这可以是FIFO或FAIR，用于控制池中的作业是否相互排队（默认）或公平分享池的资源。
• weight：这将控制池相对于其他池的池资源权重。默认情况下，所有池的权重均为1.如果给某个特定池的权重为2，则它将获得其他活动池的两倍多的资源。设置较高的权重（如1000）也可以实现 池之间的优先级 - 实质上，权重1000池将在任务激活时首先启动任务。
• minShare：除了总体权重之外，每个池还可以给予管理员希望拥有的最小份额（作为CPU核心数量）。公平调度程序总是试图满足所有活动池的最小份额，然后根据权重重新分配额外资源。minShare因此，该属性可以是另一种确保池总能快速获得特定数量的资源（例如10个内核）的另一种方式，而不会为群集的其余部分提供高优先级。默认情况下，每个池minShare都是0。

可以通过创建XML文件来设置池属性，在类路径下的fairscheduler.xml文件 （conf/fairscheduler.xml.template），或者spark.scheduler.allocation.file在SparkConf中设置文件路径 。

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

XML文件中，标签<pool>表示每个池，其中有不同的元素用于各种设置。例如：

<?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>

一个完整的例子也可以在conf/fairscheduler.xml.template看到。请注意，任何未在XML文件中配置的池都将使用默认值（调度模式FIFO，权重1和minShare 0）。