Spark Job Scheduling

绝地灬酷狼 2023-06-19 11:14 3阅读 0赞

**Table of Contents**

跨程序调度

动态资源分配

配置和设置

资源分配策略

安全移除executor

程序内调度

公平调度池

调度池的默认行为

调度池配置

--------------------

Spark 有几个用于在计算之间调度资源的工具。首先，回想一下，正如集群模式概述中所述，每个 Spark 应用程序(SparkContext的实例)运行一组独立的执行器进程。Spark 所运行的集群管理器为跨应用程序调度提供了便利。其次，在每个 Spark 应用程序中，如果由不同的线程提交多个“作业”( Spark Actions)，那么它们可能并发运行。如果程序正在通过网络服务请求，这是很常见的。Spark 包含一个公平调度程序调度每个 SparkContext 中的资源。

# 跨程序调度 #

在集群上运行时，每个Spark应用程序都有一组独立的 executor jvm，它们只运行任务并存储应用程序的数据。如果多个用户需要共享集群，根据集群管理器的不同，可以使用不同的选项来管理分配。

所有集群管理器上都可以使用的最简单的选项是资源的静态分区。通过这种方法，每个应用程序都被赋予了它可以使用的最大资源量，并在整个过程中一直保留这些资源。这是 Spark 的独立模式 和 YARN 模式以及粗粒度 Mesos 模式中使用的方法。根据集群类型，可以配置如下资源分配：

*  **Standalone mode - **默认情况下，提交给独立模式集群的应用程序将以 FIFO(先进先出)顺序运行，每个应用程序将尝试使用所有可用的节点。可以通过设置 spark.core.max 来限制应用程序使用的节点数量。或者更改没有通过 spark.deploy. defaultCores 设置此设置的应用程序的默认值。最后，除了控制 cores 之外，还可以通过设置 spark.executer.memory 来控制他们使用的内存。
 *  **Mesos - **要在Mesos上使用静态分区，设置 spark.mesos.coarse 为 true，通过可选项 spark.cores.max 来限制单机模式下每个程序的资源。还应该设置 spark.executer.memory 来控制内存。
 *  **YARN - **`--num-executors 选项控制 Spark 客户端 executer 的数量，--executor-memory 和 --executor-cores 控制每个 executer 的资源。`[Running Spark on YARN][]

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

注意，目前没有一种模式提供跨应用程序的内存共享。如果您希望以这种方式共享数据，我们建议运行一个服务器应用程序，它可以通过查询相同的 RDDs 来服务多个请求。

## 动态资源分配 ##

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

该功能在默认情况下是禁用的，在所有粗粒度集群管理器(即独立模式、YARN 模式和 Mesos 粗粒度模式)上都可用。

### 配置和设置 ###

使用这个特性有两个要求。首先，应用程序必须设置 `spark.dynamicAllocation.enabled = true，`其次，必须在同一集群中的每个工作节点上设置一个外部 shuffle 服务，并设置 `spark.shuffle.service.enabled = true，`外部 shuffle 服务的目的是允许删除执行者而不删除由他们写的洗牌文件(更多细节将在下面描述)。设置此服务的方式因集群管理器而异：

*  在独立模式下，只需使用 `spark.shuffle.service.enabled` = `true`
 *  Mesos 粗粒度模式下，所有子节点运行 `$SPARK_HOME/sbin/start-mesos-shuffle-service.sh，spark.shuffle.service.enabled` = `true`
 *  YARN 模式，查看：[Running Spark on YARN][]

所有其他相关的配置都是可选的，`spark.dynamicAllocation.*` 和 `spark.shuffle.service.*，更多查看：`[Spark Configuration Guide][]

### 资源分配策略 ###

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

**请求策略**

具有动态分配的Spark应用程序在有等待调度的挂起任务时，会启用额外的执行器请求。这个条件必然意味着现有的执行者集不足以同时满足所有已提交但尚未完成的任务。

Spark 以轮为单位请求执行者。当 spark.dynamicAllocation.`schedulerBacklogTimeout `存在挂起的任务时，将触发实际的请求。如果挂起的任务队列持续存在，然后每隔 `spark.dynamicAllocation.sustainedSchedulerBacklogTimeout 秒之后，`此外，每轮申请的 executor 的数量比前一轮呈指数级增长。例如，一个应用程序将在第一轮中添加一个executor，然后在随后的几轮中添加2、4、8等 executor。

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

**移除策略**

删除执行者的策略要简单得多。当执行程序空闲超过 `spark.dynamicAllocation.executorIdleTimeout` seconds 时，程序移除 executor。请注意，在大多数情况下，此条件与请求条件是互斥的，因为如果仍然有有待调度的任务，则执行程序不应该空闲。

## 安全移除executor ##

在动态分配之前，Spark 执行程序在失败时或关联的应用程序退出时退出。在这两个场景中，不再需要与执行程序关联的所有状态，可以安全地丢弃它们。然而，使用动态分配时，当显式删除执行程序时，应用程序仍在运行。如果应用程序试图访问存储在执行程序或由执行程序编写的状态，则必须重新计算状态。因此，Spark需要一种机制，通过在删除执行者之前保留其状态来安全地使其退役。

这一要求对 shuffle 尤其重要。在洗牌期间，Spark 执行器首先将自己的映射输出本地写到磁盘，然后当其他执行器试图获取这些文件时充当这些文件的服务器。对于运行时间比其他执行器长得多的任务，动态分配可能会在 shuffle 完成之前删除执行器，在这种情况下，由该执行器编写的 shuffle 文件必须进行不必要的重新计算。

保存 shuffle 文件的解决方案是使用外部 shuffle 服务，这也是在Spark 1.2中引入的。此服务指的是在集群的每个节点上独立于Spark 应用程序及其执行者运行的长时间运行的流程。如果服务被激活，Spark 执行器将从服务中获取 shuffle 文件，而不是从其他服务中获取。这意味着由执行程序编写的任何 shuffle 状态都可以在执行程序的生命周期之外继续执行。

除了写 shuffle 文件，执行者还将数据缓存在磁盘或内存中。然而，当删除执行程序时，所有缓存的数据将不再可访问。为了缓解这个问题，默认情况下包含缓存数据的 executor 永远不会被删除。也可以使用 spark.dynamicAllocation.cachedExecutorIdleTimeout 配置此行为。在未来的版本中，缓存的数据可以通过一个非堆存储来保存，这在本质上类似于通过外部 shuffle 服务来保存 shuffle 文件。

# 程序内调度 #

在给定的Spark应用程序(SparkContext实例)中，如果多个并行作业从不同的线程提交，它们可以同时运行。在本节中，我们所说的“作业”指的是一个Spark action(例如，save、collect)和需要运行以评估该操作的任何任务。Spark 的调度器是完全线程安全的，并且支持这个用例来启用服务多个请求的应用程序(例如，针对多个用户的查询)。

默认情况下，Spark 的调度程序以 FIFO 方式运行作业。每个作业被划分为“stage”(例如map和reduce阶段)，第一个作业获得所有可用资源的优先级，而它的阶段有要启动的任务，然后第二个作业获得优先级，等等。如果队列头部的作业不需要使用整个集群，则稍后的作业可以立即开始运行，但如果队列头部的作业很大，则稍后的作业可能会显著延迟。

从Spark 0.8开始，还可以配置作业之间的公平共享。在公平共享下，Spark以“轮询”的方式在作业之间分配任务，以便所有作业获得大致相等的集群资源共享。这意味着在长作业运行时提交的短作业可以立即开始接收资源，并且仍然可以获得良好的响应时间，而不必等待长作业完成。此模式最适合多用户设置。

要启用公平调度程序，只需设置

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

## 公平调度池 ##

公平调度程序还支持将作业分组到池中，并为每个池设置不同的调度选项(如权重)。这对于创建更重要的作业的“高优先级”池很有用，例如，或者将每个用户的作业分组在一起，并给用户平等的共享，不管他们有多少并发作业，而不是给作业平等的共享。此方法模仿 Hadoop Fair 调度程序。详见：[Hadoop: Fair Scheduler][Hadoop_ Fair Scheduler]

在没有任何干预的情况下，新提交的作业将进入默认池，但是可以通过添加 spark.scheduler.pool 来设置作业池：

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

在设置了这个本地属性之后，在这个线程中提交的所有作业(通过在这个线程中调用 `RDD.save`, `count`, `collect `等)将使用此池名称。设置为每个线程，以便让一个线程很容易地代表同一个用户运行多个作业。如果你想清除线程关联的池，只需调用：

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

## 调度池的默认行为 ##

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

## 调度池配置 ##

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

*  `schedulingMode - `FIFO or FAIR，控制池内的作业是否在彼此之间排队(默认)或公平地共享池的资源。
 *  `weight - `这控制池在集群中相对于其他池的份额。默认情况下，所有池的权值为1。例如，如果为一个特定的池赋予2的权重，它将获得比其他活动池多两倍的资源。设置一个像1000这样的高权重也使得在池之间实现优先级成为可能——本质上，weight-1000池总是在作业激活时首先启动任务。
 *  `minShare - `除了总体权重之外，每个池还可以获得管理员希望它拥有的最小共享(作为 CPU 内核的数量)。公平调度程序总是试图在根据权重重新分配额外资源之前满足所有活动池的最小份额。因此，minShare 属性可以作为另一种方式来确保一个池总是能够快速地获得一定数量的资源(例如10个核)，而不会为集群的其他部分提供高优先级。默认情况下，每个池的minShare为0。

可以通过创建一个类似于 `conf/fairscheduler.xml.template `文件来设置池属性。并设置spark.scheduler.allocation.file 属性。

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)。

原文地址：[http://spark.apache.org/docs/2.1.1/job-scheduling.html\#overview][http_spark.apache.org_docs_2.1.1_job-scheduling.html_overview]

[Running Spark on YARN]: https://blog.csdn.net/zpf_940810653842/article/details/103377694
[Spark Configuration Guide]: https://blog.csdn.net/zpf_940810653842/article/details/103372744
[Hadoop_ Fair Scheduler]: https://blog.csdn.net/zpf_940810653842/article/details/103566426
[http_spark.apache.org_docs_2.1.1_job-scheduling.html_overview]: http://spark.apache.org/docs/2.1.1/job-scheduling.html#overview