Spark文档 - 作业调度

概述

Spark提供了几个可以在计算过程之间调度资源的工具。首先，每个Spark应用程序（SparkContext实例）都运行在独立的executor进程中，而集群管理器可以跨应用程序调度资源。其次，Spark应用程序内部的众多“作业”可能会并行执行，这种场景在处理网络请求时很常见。Spark包含了一个公平调度器在SparkContext内部调度资源。

跨应用程序调度

Spark应用程序在集群上运行时都有独立的executor进程负责执行任务和存储数据，如果多个用户想要共享集群，不同的集群管理器有不同的做法。

最简单的方式是对资源静态分区，适用于所有集群管理器。这种方式会为应用程序分配最大可用资源，应用程序会在整个运行期间都持有并使用这些资源。这个方式通常用在Standalone，YARN和coarse-gained Mesos模式下。资源可以通过如下方式分配：

• Standalone：该模式下，应用程序以FIFO顺序执行，每个应用程序都会尝试使用所有的资源。可以在应用程序内部设置spark.cores.max来限制可用CPU数，或者修改配置文件中的spark.deploy.defaultCores选项来更改默认CPU数。除了控制CPU，应用程序还可以通过spark.executor.memory控制内存。
• Mesos：在Mesos上使用静态分区需要设置spark.mesos.coarse为true。之后使用spark.cores.max和spark.executor.memory控制每个应用程序可用的资源。
• YARN：--num-executors用于控制分配给每个应用程序的executor数。--executor-memory和--executor-cores用于控制每个executor可以使用的资源。

第二种方式是在Mesos模式下动态共享CPU。这种模式下，每个应用程序依然拥有固定且独立的内存（使用spark.executor.memory设置），但是当应用程序不执行任务时，其他应用程序可能使用空闲的CPU执行任务。这种模式适用于拥有大量非活跃应用程序的场景，例如不同用户使用的shell会话。当然，它也可能带来一定的延迟，因为在应用程序之间协调CPU需要一点时间。使用mesos://并将spark.mesos.coarse为false即可启用该模式。

需要注意的是，Spark暂不支持内存共享。

动态资源分配

Spark可以根据工作负载动态调整应用程序使用的资源。这意味着应用程序可以在不使用时将资源还给集群并在必要时重新请求资源。这一功能在多个应用程序共享资源的时候很有用。

这一功能默认未开启。

配置和设置

开启该功能需要两步：首先设置spark.dynamicAllocation.enabled为true，其次在集群中每个worker节点上设置一个外部洗牌服务并在应用程序中配置spark.shuffle.service.enabled为true。使用外部洗牌服务的目的是为了让executor移除时不必删除与之关联的洗牌文件。以下是设置该服务的方法：

• Standalone：设置spark.shuffle.service.enabled为true。
• Mesos：在所有worder节点上运行$SPARK_HOME/sbin/start-mesos-shuffle-service.sh，还要设置spark.shuffle.service.enabled为true。
• YARN：要在每个NodeManager上启动外部洗牌服务，需要以下步骤：
  1. 找到spark-<version>-yarn-shuffle.jar文件，一般可以在$SPARK_HOME/yarn文件夹下找到。
  2. 将jar包添加到所有NodeManager节点的类路径中。
  3. 在每个节点的yarn-site.xml中，设置yarn.nodemanager.aux-services的值为spark_shuffle，之后设置yarn.nodemanager.aux-services.spark_shuffle.class为org.apache.spark.network.yarn.YarnShuffleService。
  4. 通过编辑etc/hadoop/yarn-env.sh文件中YARN_HEAPSIZE属性，增加NodeManager的堆空间以避免洗牌期间的垃圾收集问题。
  5. 重启集群中所有的NodeManager。

其他相关的配置项都在spark.dynamicAllocation.*和spark.shuffle.service.*命名空间之下。

资源分配策略

从较高的层次来看，Spark应当在不使用时归还executor，并在必要时获取executor。由于没有一个明确的方法来预测一个即将被移除的executor是否会在不久之后执行任务，或者即将添加的executor是否是空闲的。我们需要一点启发式的方式来确定什么时候移除和请求executor。

请求策略

启用了动态分配的应用程序会在任务堆积时请求额外的executor。这个条件说明现有的executor无法有效的处理完提交的任务。

Spark按轮次请求executor。当任务调度延迟时间超过spark.dynamicAllocation.schedulerBacklogTimeout秒时会触发请求，之后每隔spark.dynamicAllocation.sustainedSchedulerBacklogTimeout秒后，如果任务依然堆积会再次触发请求。每隔轮次请求的executor数量按指数递增。例如，应用程序会在第一次请求新增1个executor，之后每个轮次会请求2，4，8个executor。

使用指数递增策略的动机有两方面。首先，应用程序在开始请求executor时应当谨慎一点，因为可能少量executor就足够了。其次，如果发现确实需要更多executor，应用程序也可以及时的提高使用的资源。

移除策略

移除executor的策略相对简单。空闲时间超过spark.dynamicAllocation.executorIdleTimeout的executor会被应用程序移除。在多数情况下，这个条件和请求条件是互斥的。

优雅的停用executor

不使用动态分配时，executor会在执行失败或者应用程序退出时退出。这两种情况下，与executor关联的状态都可以安全的废弃。但是使用了动态分配后，应用程序显式移除某个executor后依然运行。这时如果应用程序尝试访问与被移除executor相关联（存储或者写入）的状态，Spark就不得不重新执行该状态。因此，Spark需要一种机制，通过在移除executor之前保留其状态来优雅的停用executor。

这种机制对洗牌来说很重要。在洗牌期间，executor首先将自己的映射数据输出到本地磁盘，然后作为服务器供其它executor获取这些输出文件。如果某个事件掉队了，即某个任务的执行时间比同批次任务的时间更长，这时动态分配机制可能会在洗牌完成前就移除了executor，这样与被移除executor关联的洗牌文件就必须重新计算，这时不必要的。

保留洗牌文件的解决方案是使用外部洗牌服务，该服务是一个运行在所有worker节点上的，独立于应用程序和executor的进程。如果启用了该服务，executor就会从该服务获取洗牌文件。这样executor的状态就可以超越executor的生命周期继续保留。

除了洗牌文件，executor还会在内存或磁盘上缓存数据。executor移除后，这些缓存数据也无法访问了。为了缓解这种情况，包含缓存数据的executor是不会被移除的。可以通过spark.dynamicAllocation.cachedExecutorIdleTimeout选项配置。

应用程序内调度

在应用程序内部，多个并行作业可以同时执行。这里的“作业”指的是Spark action（例如save和collect）以及能推导出action的任务。Spark的调度器是线程安全的，允许应用程序处理多个请求。

Spark调度器默认以FIFO顺序执行作业。每个作业分割成多个“stage”（例如map和reduce阶段）。第一个作业首先获得使用资源的优先级并执行任务，之后是第二个作业，以此类推。如果队列中第一个作业不需要使用整个集群，那么后续作业也可以立即执行。但是如果作业很大，那么后续作业会明显的被延迟。

可以为作业配置一个公平共享策略。这种策略下，Spark会以“循环”风格为作业分配任务，这样左右的作业都可以得到大致相等的资源。这意味着小型作业不必等待大型作业执行完毕也可以执行，从而得到较好的响应时间。

要启用公平调度器，只需：

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

公平调度器池

公平调度器可以将作业分组成池，每个池可以设置不同的调度选项。这样可以为更重要的作业创建一个“高优先级”池，或者是将每个用户的作业分组到一起，然后为用户分配资源。

不使用任何配置，新提交的作业属于默认池。可以使用spark.scheduler.pool选项为作业分配池：

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

设置完毕后，所有由该线程提交的作业都会使用这个池。设置是线程级别的，如果想移除池，只需要：

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

池的默认行为

每个池默认平分集群资源，但在池内部，作业按照FIFO顺序执行。

配置池属性

池属性可以通过配置文件修改，池支持三个属性：

• schedulingMode：可以是FIFO和FAIR，控制池内部作业的执行情况。
• weight：控制池相对于其它池的权重。默认为1。如果给某个池的权重是2，那么这个池会得到其它池两倍的资源。
• minShare：除了权重，还可以为每个池分配最小资源量（例如CPU核心数）。

池属性通过XML文件设置，模板在conf/fairscheduler.xml.template中。需要新建fairscheduler.xml文件并放置到类路径中，或者通过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>