Francisco Oliveira 是 AWS _专业服务团队的顾问_客户开始大数据之旅时，通常会需要一些关于如何将用户应用程序提交到
Amazon EMR 的
Spark 方面的指导。例如，客户会需要以下方面的指导：如何设定可用于其应用程序的内存和计算资源的大小，以及适用于其使用案例的最佳资源分配模型。在本文中，我将介绍如何通过设置
spark-submit 标志来控制提交到在 EMR 上运行的 Spark 应用程序可用的内存和计算资源。我还会探讨在哪些情况下使用 maximumResourceAllocation 配置选项和executor执行程序的动态资源分配。

Spark 执行模型

从较高维度看，每个Spark应用程序都有一个driver驱动程序，它以任务形式为集群多个节点上运行的executor执行程序分配工作。

Driver驱动程序是一种应用程序代码，用于定义应用于数据集的转换和动作。driver驱动程序的核心是实例化的 SparkContext 类的对象。该对象允许driver驱动程序获取到集群的连接、请求资源、将应用程序操作拆分为任务，以及在executor执行程序中安排和启动任务。

Executor执行程序不仅执行驱动driver程序发送的任务，还会在本地存储数据。随着executor执行程序不断创建和销毁（请参阅后文的“启用executor执行程序的动态资源分配”部分），它们通过driver驱动程序注册和注销。Driver驱动程序和executor执行程序直接进行通信。

要执行应用程序，driver驱动程序会组织要在作业中完成的工作。每个作业分为多个阶段，每个阶段由一组并行运行的独立任务组成。任务是 Spark 中的最小工作单元，用于在不同的分区执行相同的代码。

Spark 编程模型

弹性分布式数据集 (RDD) 是 Spark 中的一个重要抽象层。这一抽象层是执行内存中计算的关键。RDD 是分布在集群节点之间的只读和不可变数据分区集合。Spark 中的分区允许并行执行数据子集。Spark 应用程序会创建 RDD 并对 RDD 执行操作。尽管 Spark 会自动对 RDD 进行分区，但您也可以自己设置分区数。

RDD 支持两种类型的操作：转换（transform）和动作(action)。转换是生成新 RDD 的操作，而动作是在数据集上运行转换后，将数据写入外部存储或将值返回driver驱动程序的操作。常见的转换包括按照键进行筛选、排序和分组。常见的动作包括收集任务结果并将其发送给driver驱动程序、保存 RDD 或计算 RDD 中的元素数。

spark-submit

在集群上启动应用程序的常用方法是使用 spark-submit 脚本。该脚本提供了几个标志，可让您控制应用程序使用的资源。

设置 spark-submit 标志是向在driver驱动程序中实例化的 SparkContext 对象动态提供配置的一种方法。当您创建集群和在应用程序中进行硬编码（但不建议这么做）时，spark-submit 还可以读取您使用 EMR 配置选项设置的 conf/spark-defaults.conf) 文件中设置的配置值_。_更改 conf/spark-defaults.conf 的另一种方法是使用 –conf prop = value 标志。我将介绍 spark-submit 标志和在 spark-defaults.conf 文件中使用的属性名称以及 –conf 标志。

在 EMR 上运行的 Spark 应用程序

提交到EMR 上的 Spark 的任何应用程序都会在 YARN 上运行，且每个 Spark executor执行程序都会作为 YARN 容器运行。在 YARN 上运行时，driver驱动程序可以在集群中的某个 YARN 容器中运行（集群模式），也可以在 spark-submit 进程中本地运行（客户端模式）。

在集群模式下运行时，driver驱动程序在 ApplicationMaster 上运行，ApplicationMaster 根据应用程序所需资源向 YARN ResourceManager 提交 YARN 容器请求。一个简化的和高维度的应用程序提交流程示意图如下所示。

在客户端模式下运行时，driver驱动程序在 ApplicationMaster 外部运行，它运行在提交应用程序的机器上的_spark-submit_ 脚本进程中。

设置driver驱动程序位置

借助 spark-submit，可以使用 –deploy-mode 标志选择driver驱动程序的位置。

如果希望进行调试并快速查看应用程序的输出时，可以选择以客户端模式提交应用程序。对于生产环境的应用程序，最佳实践是在集群模式下运行应用程序。此模式可确保在应用程序执行期间driver驱动程序始终可用。但是，如果您使用客户端模式，并且是在 EMR 集群外部（例如，在本地的笔记本电脑上）提交应用程序，需要注意driver驱动程序是在 EMR 集群外部运行，driver驱动程序与executor执行程序之间的通信延迟会增加。

设置driver驱动程序资源

Driver驱动程序的大小取决于它执行的计算量，以及它从executor执行程序收集的数据量。在集群模式下运行driver驱动程序时，spark-submit 为您提供了相关选项，可控制driver驱动程序使用的核心数 (–driver-cores) 和内存 (–driver-memory)。在客户端模式下，driver驱动程序内存的默认值为 1024MB 和1个核心。

设置executor核心数和执行程序数

Executor执行程序核心数_（__–executor-cores_ 或 spark.executor.cores）确定了每个executor执行程序可以并行执行的任务数。最佳实践是：为操作系统保留1个核心，为每个executor执行程序保留大约 4-5 个核心。请求的核心数受配置属性 yarn.nodemanager.resource.cpu-vcores 限制，该属性可以控制在一个节点中运行的所有 YARN 容器可用的核心数，该属性可以在 yarn-site.xml 文件中设置。

可以使用以下公式计算每个节点的executor执行程序数：

每个节点的__executor执行程序数 = （节点的核心数 – 为操作系统保留的 1 个核心）__/ 每个__executor执行程序的任务数

Spark 作业的executor执行程序_（__—_num-executors 或 spark.executor.instances）总数为：

executor执行程序总数 = 每个节点的executor执行程序数 * 实例数 – 1。

设置executor执行程序的内存

每个executor执行程序容器的内存空间都分为两个主要区域：Spark executor执行程序内存和overhead内存。

请注意：可以分配给executor执行程序容器的最大内存取决于 yarn-site.xml 中可用的 yarn.nodemanager.resource.memory-mb 属性。executor执行程序内存（–executor-memory 或 spark.executor.memory）确定了每个executor执行程序进程可以使用的内存量。overhead内存 (spark.yarn.executor.memoryOverHead) 是堆外内存，会自动添加到executor执行程序内存，它的默认值是 executorMemory * 0.10。

Executor执行程序内存将用于存储和执行目的堆的各个部分进行了统一。如果超出使用量，这两部分可以彼此借用空间。相关属性是 spark.memory.fraction 和 spark.memory.storageFraction。有关更多信息，请参阅 Spark 1.6 中的统一内存管理白皮书。

可以使用以下公式计算每个executor执行程序的内存：

每个executor执行程序的内存 = 节点上最大容器大小 / 每个节点的executor执行程序数

spark-submit 示例

为了向您展示如何设置之前介绍的标志，我会提交一个 wordcount 示例应用程序，然后使用 Spark history server获取执行的视图。

首先，以 EMR 步骤的形式，向现有集群提交一个修改版的 word count 示例应用程序。代码如下所示：

Bash

from __future__ import print_function
from pyspark import SparkContext
import sys
if __name__ == "__main__":
    if len(sys.argv) != 3:
        print("Usage: wordcount  ", file=sys.stderr)
        exit(-1)
    sc = SparkContext(appName="WordCount")
    text_file = sc.textFile(sys.argv[1])
    counts = text_file.flatMap(lambda line: line.split(" ")).map(lambda word: (word, 1)).reduceByKey(lambda a, b: a + b)
    counts.saveAsTextFile(sys.argv[2])
    sc.stop()

集群具有 6 个 m3.2xlarge 核心实例和 1 个主实例，每个实例有 8 个 vCPU 和 30GB 内存_。_此实例类型的 yarn.nodemanager.resource.memory-mb 默认值是 23GB。

根据上述介绍的公式，spark-submit 命令如下所示：

Bash

spark-submit --deploy-mode cluster --master yarn --num-executors 5 --executor-cores 5 --executor-memory 20g –conf spark.yarn.submit.waitAppCompletion=false wordcount.py s3://inputbucket/input.txt s3://outputbucket/

通过以下命令以 EMR 步骤的形式提交应用程序：

Bash

aws emr add-steps --cluster-id j-xxxxx --steps Type=spark,Name=SparkWordCountApp,Args=[--deploy-mode,cluster,--master,yarn,--conf,spark.yarn.submit.waitAppCompletion=false,--num-executors,5,--executor-cores,5,--executor-memory,20g,s3://codelocation/wordcount.py,s3://inputbucket/input.txt,s3://outputbucket/],ActionOnFailure=CONTINUE

请注意：还可通过步骤定义设置属性 spark.yarn.submit.waitAppCompletion。当此属性设置为 false 时，客户端提交应用程序然后就可以退出，不需要等待应用程序完成。此设置可让您提交多个应用程序由集群同时执行，但是它仅在集群模式下可用。

使用 –driver-memory 和 –driver-cores 的默认值，因为示例应用程序将直接写入 Amazon S3，并且driver驱动程序没有从executor执行程序接收任何数据。

启用executor执行程序的动态资源分配

YARN 上的 Spark 可以动态扩展和缩减executor执行程序的数量。该功能非常有用，在同时处理多个应用程序时可以释放空闲executor执行程序，让某个应用程序可以按需请求其他executor执行程序。

要启用该功能，请参阅 EMR 文档，了解相关步骤。

Spark 具有以下属性，可为动态分配机制提供精细控制：

executor执行程序的初始数量 (spark.dynamicAllocation.initalExecutors)
应用程序使用的最小executor执行程序数 (spark.dynamicAllocation.minExecutors)
可以请求的最大executor执行程序数 (spark.dynamicAllocation.maxExecutors)
删除空闲executor执行程序的时间 (sparkdynamicAllocation.executorIdleTime)
请求新executor执行程序处理等待任务的时间（spark.dynamicAllocation.schedulerBacklogTimeout 和 spark.dynamicAllocation.sustainedSchedulerBacklogTimeout）

使用最大资源分配自动配置executor执行程序

EMR 提供了一个能够自动配置这个属性的选项，以实现整个集群资源利用率最大化。如果集群一次只处理一个应用程序，此配置选项会非常有用。如果您希望同时运行多个应用程序，请不要使用该选项。

要启用该配置选项，请参阅 EMR 文档，了解相关步骤。

在集群创建期间设置此配置选项后，EMR 会自动更新 spark-defaults.conf 文件，来控制executor执行程序计算和内存资源的属性，如下所示：

spark.executor.memory = (yarn.scheduler.maximum-allocation-mb – 1g) -spark.yarn.executor.memoryOverhead
spark.executor.instances = [将此设置为核心节点的初始数量加上任务节点数]
spark.executor.cores = yarn.nodemanager.resource.cpu-vcores
spark.default.parallelism = spark.executor.instances * spark.executor.cores

查看并行任务视图

您可以通过 EMR 控制台访问Spark history server UI。它提供了有关应用程序性能和行为的有用信息。您可以查看计划的阶段和任务列表、检索有关executor执行程序的信息、获取内存使用情况摘要，以及检索提交到 SparkContext 对象的配置。在本文中，我将介绍如何将以上示例中使用的 spark-submit 脚本设置的标志转换为可视化工具。

要访问 Spark history server，请启用 SOCKS 代理，然后在连接下选择 Spark History Server。

对于Completed applications，请选择唯一可选的条目，然后展开事件时间轴，如下所示。Spark 根据_–num-executors_ 标志中定义的要求添加了 5 个executor执行程序。

然后，导航到阶段详细信息，您就可以看到每个executor执行程序并行运行的任务数了，该值与 –executor-cores 标志的值相同。

小结

本文向您介绍了如何使用 spark-submit 标志向集群提交应用程序。我们详细阐述了如何控制driver驱动程序运行位置、如何设置分配给driver驱动程序和executor执行程序的资源以及executor执行程序数量。此外，我们还介绍了在哪些情况下需使用 maximumResourceAllocation 配置选项和executor执行程序的动态资源分配。

如果您有任何问题或建议，欢迎留言。

—————————-

想要了解有关大数据或流数据的更多信息？请查看我们的大数据和流数据教学页。

本文转载自AWS技术博客。

原文链接：https://amazonaws-china.com/cn/blogs/china/submitting-user-applications-with-spark-submit/

Spark 执行模型

Spark 编程模型

spark-submit

在 EMR 上运行的 Spark 应用程序

设置driver驱动程序位置

设置driver驱动程序资源

设置executor核心数和执行程序数

设置executor执行程序的内存

spark-submit 示例

启用executor****执行程序的动态资源分配

使用最大资源分配自动配置****executor执行程序

查看并行任务视图

小结

启用executor执行程序的动态资源分配

使用最大资源分配自动配置executor执行程序

创作场景

使用 spark-submit 提交用户应用程序