Spark的作业调度机制

迷南。 2024-04-17 05:24 82阅读 0赞

1.  ##### 简介 #####
    
    Spark调度机制可以理解为两个层面的调度。Spark Application调度(Spark应用程序在集群中运行的调度,包括Driver调度和Executor调度)和单个Spark应用程序SparkContext的内部调度。
    
    SparkContext内部调度就是每个Spark Application都会有若干Jobs(Spark Actions)，然后这些job是以何种机制在Executor上执行的，也是需要一个调度管理的机制，该层面调度可以理解为SparkContext内部调度。若干 Jobs 由不同的线程提交，它们可以同时运行，此时可以使用 Spark Application 的 Fair 机制的调度资源。
2.  ##### Applications间的调度 #####
    
    在集群上运行时，每个Spark应用程序都获得一组独立的 executor jvm，这些jvm只运行该应用程序的 tasks 和缓存该应用程序的数据。当有多个应用或者多个程序在你的集群中运行时，根据集群管理器的不同，可以使用不同的选项来管理分配。
    
    最简单的方式是提供静态资源分配，也就是给每个应用程序分配固定资源，资源数在该程序整个运行期间都不会变动。这种方式出现在Spark的 Standalone, YARN 和粗粒度 Mesos 模式。资源的分配方式，在每种集群运行模式中有所不同：
    
    1.  ###### standalone模式 ######
        
        默认情况下，提交到独立模式群集的应用程序将以FIFO(先进先出)方式运行，每个应用程序将尝试使用所有可用节点。 您可以通过 `spark.cores.max` 配置属性来限制应用程序使用的节点数，或者通过 `spark.deploy.defaultCores` 更改未设置此应用程序的默认值。最后，除了控制 cores 外，每个应用程序可以使用 `spark.executor.memory` 配置控制其内存使用。
    2.  ###### Mesos ######
        
        要在Mesos上使用静态分区，请将spark.mesos.coarse配置属性设置为true，并可选择像Standalone模式一样设置 `spark.cores.max` 来限制每个应用程序的资源。您还应该设置 `spark.executor.memory` 来控制执行程序内存。
    3.  ###### YARN ######
        
        YARN客户端的 `--num-executors Spark` 选项控制在集群上分配的Executor数量（`spark.executor.instances`配置属性），而 `--executor-memory` （spark.executor.memory配置属性）和 `--executor-cores`(`spark.executor.cores`配置属性)控制每个执行程序的资源。
    
    在Mesos模式下还有一个可选项是动态共享 Core 。在这种模式下，  
    每个Spark应用程序仍然拥有固定数量和独立的内存(spark.executor.memory设置)，但是当Spark 应用程序在一个机器上没有运行的 task 的时候，其它的应用程序可以使用这些 cores 运行 tasks 。当你需要运行大量不活跃的 Spark 应用程序(例如来自不同用户的shell会话)时，此模式非常有用。但是此模式存在延迟可预测性较低的风险，因为当Spark App需要恢复使用这些cores的时候，需要等待一些时间才能使用这些core去执行任务。要使用此模式，只需要使用 mesos://URL并将spark.executor.coarse设置为false。
    
    请注意，目前没有一种模式可以跨应用程序提供内存共享。如果您希望以这种方式共享数据，我们建议运行单个服务器应用程序，通过查询相同的RDD来提供多个请求。
    
    **注意** :  
    目前没有一种模式可以跨应用程序提供内存共享。如果想进行跨应用程序共享数据，建议运行单个服务器应用程序，通过查询相同的RDD来提供多个请求或使用第三方存储，例如 Tachyon(是一个高性能、高容错、基于内存的开源分布式存储系统)来实现内存共享。
3.  ##### 动态资源分配 #####
    
    Spark提供了一种可根据工作负载动态调整应用程序占用资源的机制。这意味着如果不再使用应用程序，您的应用程序可能会将资源返回给群集，并在需要时再次请求它们。如果多个应用程序共享Spark群集中的资源，则此功能特别有用。
    
    默认情况下此功能禁用，并且可在所有粗粒度集群管理器上使用，即 Standalone，YARN模式和 Mesos粗粒度模式。
    
    1.  ###### 配置 ######
        
        使用此功能有两个要求。
        
        第一，Spark 应用程序必须设置 `spark.dynamicAllocation.enabled` 为 `true`。
        
        第二，必须在同一集群中的每个工作节点上设置外部shuffle服务，并设置 `spark.shuffle.service.enabled` 为 true。外部shuffle服务的目的是允许删除执行程序而不删除由它们写入的 shuffle 文件。
        
        不同的集群管理器，该服务的设置方式有所不同：
        
        1.  **Standalone 模式**
            
            只需在 `spark.shuffle.service.enabled` 设置为 true 的情况下启动 Worker 即可。
        2.  **粗粒度 Mesos 模式**
            
            在 `spark.shuffle.service.enabled` 设置为 true 的所有从属节点上，运行 $SPARK\_HOME/sbin/start-mesos-shuffle-service.sh 脚本
        3.  **YARN 模式**
            
            按如下所示在每个 NodeManager 上启动 shuffle 服务:
            
            1.  在编译 Spark 的时候要添加 YARN 配置。如果已经添加该属性，并分发到集群中，跳过此步骤。
            2.  找到 `spark-<version>-yarn-shuffle.jar` ,如果是你自己编译的 Spark 该 jar 应该在 `$SPARK_HOME/common/network-yarn/target/scala-<version>` 目录下，如果使用的是一个发行版，则应该在 yarn 下面。
            3.  将此 jar 添加到集群中所有的 NodeManager 的 Classpath 下。
            4.  在每个节点的 `yarn-site.xml` 上，给属性 `yarn.nodemanager.aux-services` 增加一个 `spark_shuffle` 值，然后将 `yarn.nodemanager.aux-services.spark_shuffle.class` 设置为 `org.apache.spark.network.yarn.YarnShuffleService` 。
            5.  通过设置 NodeManager 节点的 `etc/hadoop/yarn-env.sh` 文件中的 `YARN_HEAPSIZE`(默认为 1000)来增加堆大小，以避免在shuffle期间出现垃圾收集问题。
            6.  重启集群中的所有 NodeManager 。
                
                当shuffle服务在YARN上运行时，可以使用以下额外配置选项：
                
                <table> 
                 <thead> 
                  <tr> 
                   <th align="center">参数</th> 
                   <th align="center">默认</th> 
                   <th align="center">含义</th> 
                  </tr> 
                 </thead> 
                 <tbody> 
                  <tr> 
                   <td align="center">spark.yarn.shuffle.stopOnFailure</td> 
                   <td align="center">false</td> 
                   <td align="center">是否在Spark Shuffle Service初始化失败时停止NodeManager。可以防止在NodeManagers上运行 Containers 却未运行Spark Shuffle Service 而导致的应用程序故障。</td> 
                  </tr> 
                 </tbody> 
                </table>
    2.  ###### 资源分配策略 ######
        
        在较高的层面上，Spark 应该在不再使用执行程序时放弃执行程序，并在需要时获取执行程序。由于没有确定的方法可以预测即将被删除的执行程序是否会在不久的将来运行任务，或者即将添加的新执行程序实际上是否是空闲的，所以需要一组启发式来确定何时删除或请求 executors。
        
        1.  **请求策略**
            
            当启用动态分配的 Spark 应用程序有等待调度的挂起任务时，它会请求额外的执行器。这意味着现有的 Executors 不足以同时满足所有已提交但尚未完成的任务。
            
            Spark会轮询申请资源。如果挂起的任务持续 `spark.dynamicAllocation.schedulerBacklogTimeout` 数秒时会触发实际请求，如果待处理任务队列持续存在，则会在此后每隔 `spark.dynamicAllocation.sustainedSchedulerBacklogTimeout` 秒将再次触发。此外，每轮请求的 Executors 数量与上一轮相比会呈指数增长。例如，一个应用程序将在第一轮中添加1个 Executor ，然后在后续轮次中添加 2,4,8 等个 Executor。
            
            指数增长政策的动机是双重的。首先，应用程序应该在开始时谨慎地申请 Executor，以防只有几个额外的 Executor 就足够了。这与TCP慢启动的理由相呼应。其次，应用程序应该能够及时提高其资源使用率，以防实际需要许多执行程序。
        2.  **删除策略**  
            删除 Executors 的策略要简单得多。Spark 应用程序在空闲时间超过 `spark.dynamicAllocation.executorIdleTimeout` 秒时会删除 Executors。请注意，在大多数情况下，此条件与请求条件互斥，因为如果仍有待执行的任务时，则 Executor 不应处于空闲状态。
    3.  ###### Executors 优雅退出 ######
        
        在动态分配之前，Spark Executors在出现故障或退出相关应用程序时退出。在这两种情况下，与 Executors 相关联的所有状态不再需要，可以安全地丢弃。但是，通过动态分配，当显式删除 Executors 时，应用程序仍在运行。如果应用程序尝试访问由 Executors 存储或写入的状态存，则必须执行重新计算状态。因此，Spark需要一种机制，通过在删除执行程序之前保留其状态来优雅地停用 Executors。
        
        这一要求对于 Shuffle 尤其重要。在 Shuffle 期间，Spark Executor 首先将 map 输出本地写入磁盘，然后当 Executor 尝试获取这些文件时充当这些文件的服务器。如果存在运行时间比其他 Executor 长得多的任 tasks 时，动态申请的 Executor 有可能在 Shuffle 未结束之前就被移除了，在这种情况下，必须重新计算由该 Executors 写入的 Shuffle 文件。
        
        可以使用外部 shuffle 服务保存 shuffle 输出文件，从 Spark 1.2 开始引入。此服务指的是一个长期运行的进程，它独立于 Spark应用程序及其 Executors ，在集群的每个节点上运行。如果启用该服务，Spark执行程序将从服务获取shuffle文件，而不是从其它 Executor。这意味着由执行人员写入的任何 shuffle 状态可能会继续执行超出 Executor 的生命周期。
        
        除了写shuffle文件外，执行程序还可以在磁盘或内存中缓存数据。但是，删除执行程序后，所有缓存的数据将不再可访问。为了避免这种情况，默认情况下，永远不会删除包含缓存数据的 Executors。您可以使用 `spark.dynamicAllocation.cachedExecutorIdleTimeout` 配置此行为。在将来的版本中，缓存的数据可以通过堆外存储来保存，这在本质上类似于通过外部shuffle服务保存shuffle文件。
    4.  ###### Spark App 内部调度 ######
        
        在给定的Spark应用程序(SparkContext实例)中，如果从单独的线程提交多个并行作业，那么它们可以同时运行。在本节中，我们所说的 job 是指一个 Spark action (例如，save，collect)和任何需要运行的任务以评估该操作。Spark 的调度程序是完全线程安全的，并支持此用例来启用服务于多个请求的应用程序(例如针对多个用户的查询)。
        
        默认情况下，Spark的调度程序以FIFO方式运行作业。每个作业被划分为“阶段”(例如 map和reduce阶段)，第一个 job 运行结束之后，第二个job才有会去执行。如果在队列头部的job不需要使用集群的全部资源，那么后面的job可以立即执行。但是如果队列头部的 job 很大的话，其余的 job 必须推迟执行。
        
        从Spark 0.8开始，还可以配置作业之间的公平共享。在公平共享下，Spark以“循环”的方式在作业之间分配 tasks，以便所有作业获得大致相等的集群资源共享。这意味着长job运行期间提交的短工作可以立即开始接收资源，并且仍然可以获得良好的响应时间，而不必等待长 job完成。此模式最适合多用户设置。
        
        要启用公平调度程序，只需在配置 SparkContext 时将 `spark.scheduler.mode` 属性设置为 FAIR :
        
            val conf = new SparkConf().setMaster(...).setAppName(...)
            conf.set("spark.scheduler.mode", "FAIR")
            val sc = new SparkContext(conf)
        
        1.  **公平调度池**
            
            公平调度程序还支持将作业分组到池中，并为每个池设置不同的调度选项（例如: 权重）。这可以用于为更重要的job创建“高优先级”池，或者将每个用户的job组合在一起，并为用户提供相同的份额，而不管他们拥有多少并发作业，而不是给予作业相等的份额。此方法以Hadoop Fair Scheduler为模型 。
            
            在没有任何干预的情况下，新提交的作业将进入默认池，可以通过向提交线程中的SparkContext 添加 spark.scheduler.pool “local property” 来设置作业池。如下：
            
                // Assuming sc is your SparkContext variable
                sc.setLocalProperty("spark.scheduler.pool", "pool1")
            
            设置这个本地属性后，在此线程中提交所有的工作(通过此线程调用的 RDD.save，count，collect等)将使用此 pool 名称。该设置是每个线程，以便让线程代表同一个用户运行多个作业变得容易。如果要清除与线程关联的池，只需调用：
            
                sc.setLocalProperty("spark.scheduler.pool", null)
        2.  **池的默认行为**
            
            默认情况下，每个 pool 获得的群集份额相等(默认池中的每个作业的份额也相等)，但是每个池中的作业依然是FIFO的顺序运行。例如，如果您为每个用户创建一个池，这意味着每个用户将获得该群集的相等份额，并且每个用户的查询将按顺序运行，而不是以后的查询从该用户的早期查询中获取资源。
        3.  **配置池属性**
            
            可以通过配置文件修改特定池的属性。每个池支持三个属性：
            
            1.  schedulingMode  
                这可以是FIFO或FAIR，用于控制池中的作业是否以队列的形式顺序执行(默认)或公平地共享池的资源。
            2.  weight  
                这可以控制池相对于其他池的共享。默认情况下，所有池的权重均为1.例如，如果将特定池权重设置为2，则将比其他活跃池多获得2倍的资源。设置高权重(例如1000)可以实现池之间的优先级 - 实质上，只要有活动作业，weight-1000池将始终首先启动任务。
            3.  minShare  
                除了总体权重之外，每个池都可以获得管理员希望拥有的最小份额(如一些CPU内核)。公平调度程序总是尝试在根据权重重新分配额外的资源之前满足所有活动池的最小份额。因此，minShare 属性可以是另一种确保池可以始终快速达到一定数量的资源（例如10个核心）的方法，而不会为集群的其余部分赋予高优先级。默认情况下，每个池的 minShare 值为0。
        
        可以通过创建一个类似于conf/fairscheduler.xml.template 的 XML 文件来设置池属性，并在类路径中放置一个 fairscheduler.xml 文件，或者在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>
        
        conf / fairscheduler.xml.template 中提供了一个完整示例。注意，未在XML文件中配置的任何池将只获取所有设置的默认值(调度模式FIFO，权重1和minShare 0)。
    5.  ###### 使用JDBC连接进行调度 ######
        
        要为JDBC客户端会话设置Fair Scheduler池，用户可以设置spark.sql.thriftserver.scheduler.pool变量：
        
            SET spark.sql.thriftserver.scheduler.pool=accounting;

参考: http://spark.apache.org/docs/latest/job-scheduling.html