Amazon EMR 之 YARN Label 和 Amazon EC2 Spot 实例的天作之合

前言

大数据处理作为 EC2 Spot 实例推荐的使用场景之一，Amazon EMR 上的计算引擎们（包括但不限于 Hive， Spark）针对 Spot 实例做了大量优化来更好得处理中断。而本文会在上篇的 Amazon EMR 基础上继续扩展，以在 Spot 实例上运行 Spark 工作负载为例，让您了解如何更好得对 EMR 进行配置，让 Amazon EMR 和 Spot 实例合作得更愉快。在阅读本文之前，您需要对以下 AWS 服务有所了解。

1. Amazon EMR

Amazon EMR 是一个托管的集群平台，可简化在AWS上运行大数据框架（如 Apache Hadoop 和 Apache Spark）的过程，以处理和分析海量数据。用户可一键启动包含了众多 Hadoop 生态数据处理，分析相关服务的集群，而无需手动进行复杂的配置。

2. Amazon EC2 Spot 实例

Amazon EC2 Spot 是一种实例购买选项，让您可以利用 AWS 云中未使用的 EC2 容量。与按需实例的价格相比，使用 Spot 实例最高可以享受 90% 的折扣。您可以将 Spot 实例用于各种无状态、容错或者灵活的应用程序，例如大数据、容器化工作负载、CI/CD、Web 服务器、高性能计算 (HPC) 以及测试和开发工作负载。

3. 背景介绍

由于 EMR 本身部署简单，易扩展，并且可以通过 EMRFS 将数据存放在 S3 实现计算与存储的分离，目前越来越多的用户考虑在 EMR 上运行 Spark 工作负载。考虑到在运行大规模 ETL 任务同时节省成本的需求，用户一方面可以结合调度工具按需启动 EMR 临时集群，在批量任务完成之后终止集群，这样只会收取集群运行期间的费用；另一方面，用户可以将 EC2 Spot 实例作为 EMR 集群的计算节点，进一步降低计算成本。由于 Spot 实例使用的是 AWS 中EC2的空闲资源，会有一定中断的概率。因此我们需要考虑如何解决以下两个问题：1）Spot 实例的中断是不可避免的，如何在实例被回收后，能够及时补充资源满足计算要求；2）需要避免实例回收造成的 Spark 任务中断的问题。很多用户的 Spark ETL 作业持续时间长，如果无法解决该问题，就需要重启 Spark 任务，这样不仅增加了运维复杂度，也会造成不必要的成本开销。

Spot 实例中断处理

本节将会讨论如何配置 Spot 实例池，来确保有足够的资源启动 EMR 集群，以及在集群运行的过程中，当 Spot 实例发生中断，如何维持 EMR 集群的稳定性。

1. 将 Spot 实例应用于合适的 EMR 节点类型

EMR 有三种节点类型，分别是 Master Node，Core Node 和 Task Node，其中在 Master Node 上通常会运行分布式应用程序的主要组件，如 HDFS NameNode，Yarn ResourceManager等，来跟踪提交到集群的任务状态。Core Node 除了提供 HDFS 存储空间，还承担计算任务，在其上会运行 DataNode，Yarn NodeManager 等守护进程。Task No de 只负责提供计算资源，在其上只运行 Yarn NodeManager ，因此 Task Node在计算完成后可随意丢弃，而不影响集群的稳定性，利用这一特性可使得 EMR 集群能轻松地根据工作负载扩缩容。关于三种节点类型的详细说明，可以参考 Amazon EMR之EMR和Hadoop的前世今生这篇文章，本文聚焦在和 Spot 实例的搭配使用，不同节点类型的细节便不再赘述。

综上所述，如果追求集群持续平稳的运行，Master Node 就不适合使用 Spot 实例了，Spot 实例的中断会造成 Master Node 终止，因此整个集群就会变得不再可用。若用户有使用 HDFS 存储数据，则不建议在 Core Node 中使用 Spot 实例，否则会面临数据丢失的风险，同时本文后续会提到，提交 Spark 任务后，我们建议将 Yarn Application Master 运行在 Core Node上，若节点中断，也会造成执行过程中的 Spark 任务中断。

因此，考虑到一个 EMR 集群能够在一段时间稳定运行 Spark 作业的场景，最适合使用 Spot 实例的节点类型为 Task Node，因其本身只提供计算资源无状态的特性，即使节点中断也不会影响集群正常运行。在大批量数据处理的生产环境中，用户可以考虑对 Master Node 与少量所有的 Core Node 应用按需实例，若仍然需要使用 HDFS 存储数据，则可以增加 Core Node 实例上的存储空间，而不需要增加 Core Node 数量，将 Spot 实例应用在所有 Task Node 上，当 Spot 实例中断，EMR 会尝试启动新的实例补充计算资源，从而在降低成本的同时，能够保证任务执行完成。

2. Spot 实例资源选型

当我们在 EMR 集群中使用 Spot 实例，就需要考虑实例中断带来的影响，不同 Spot实例类型的中断概率不一样，我们可以通过 Spot Advisor 选择中断概率较低的实例从而降低中断概率频次。

Spot 实例的中断原因与底层资源池中 EC2 资源的丰富程度有关，我们可以选择在不同的可用区中，使用不同类型的实例以及不同大小的实例来丰富资源池。由于 EMR 集群只能在一个可用区中部署，为了在 Spot 实例中断后能及时补充资源，我们只需考虑在 Task Node 对应的实例队列或实例组中使用不同类型以及不同大小的实例即可。

例如，我们在可以在 Task Node 的实例队列中同时使用 m5a.2xlarge 与 m5.2xlarge 类型实例，当 m5.2xlarge 资源池中的实例不够用时，EMR 可以启动 m5a.2xlarge 实例作为补充；也可以同时使用 m5.2xlarge 与 m5.large 类型实例，当 1 台 m5.2xlarge 被回收，EMR 会启动两台（启动数量取决于用户根据不同类型实例所定义的 units）m5.large 实例继续提供算力。

3. 实例队列 vs 实例组

在创建 EMR 集群时，我们可以选择使用实例队列或实例组，本文总结了两种配置选项的差异：

	实例队列	实例组
混合实例类型	支持	一个实例组对应一种机型，可以定义多个实例组
混合实例购买选项	支持	一个实例组对应一种实例购买选项，可以定义多个实例组
Provisioning timeout	支持	不支持
多个子网	支持	不支持
Allocation Strategy	支持	不支持
集群运行阶段，配置文件的Reconfigure（如 yarn-site.xml 等）	不支持	支持

*Provisioning timeout：定义Spot实例在没有资源无法启动的一段时间后，转化为启动按需实例或终止集群。

*多个子网：在集群启动阶段，用户可提供多个 AZ 中的子网，EMR 会在资源相对丰富的一个 AZ 中启动集群。

*Allocation Strategy：在用户提供的多种类型 EC2 实例资源池中优先启动价格最低的按需实例与容量充足的 Spot 实例。启用该配置，可以降低 Spot 实例中断的概率。

通过表格对比我们可以观察到，由于 Provisioning timeout ，Allocation Strategy 与多个子网的特性，使用实例队列更适合运行 Spot 实例的临时 EMR 工作负载。以下给出实例队列的配置示例，来观察下如何在一个 Task Node 对应的实例队列中配置多种类型的 Spot 实例：

其中 Target capacity 为集群目标容量，units 数量决定了实例的数量，默认一个 vcpu等于一个 unit，因此可以观察到：1）Core 节点的 Target capacity 为 8 个 On-demand units，m5.xlarge 有 4 个 vcpu，因此创建集群后会启动两台按需 m5.xlarge 的 Core 节点；2）Task 节点的 Target capacity 为 12 个 Spot units，即 Task node 全部为 Spot 实例。启动集群后，EMR 会根据 Task 实例队列中定义的不同实例类型的 units 来启动实例，units 的累计值为 12 即可。因此可能会出现 3 台 m5.xlarge ，3 台 m5a.xlarge 或 1 台 m5.2xlarge 和其余 1 台机型组合的情况。若启用了 Allocation Strategy，实例队列则会从这三种实例中选择容量最优化的实例类型来启动，从而降低 Spot 实例中断的概率。

Spark 任务中断处理

Spot 实例的回收有可能会造成 Spark 任务的中断，本节将介绍如何避免该情况的发生。

1. Spark on Yarn 运行机制

当客户端向 Yarn 提交 Spark 任务后（以 cluster 的 deploy mode 为例），任务执行过程如下所示：

其中，AM（Application Master）运行在 Core 节点上，负责从 ResourceManager 协调适当的资源，跟踪运行在 Task 节点上的 Executor 执行任务的状态。假设 Task 节点使用的是 Spot 实例，当发生中断时，会发生其中一种情况：1）AM 将中断的 Executor 启动在有空闲资源其他节点上；2）若没有空闲资源，则 AM 会等待资源补充完成后，再运行 Executor。只要 AM 的进程没有中断，就可以保证 Spark 任务的稳定运行，这也是我们对 AM 所在的 Core 节点使用按需实例的原因。但是如何指定将 AM 启动在 Core 节点上，而 Executor 启动在 Task 节点上呢？这是通过 Yarn Label 来实现的，我们将在下一节进行介绍。

2. 通过 Yarn Label Exclusive 将 Core 节点设置为 AM 的独占分区

使用 Yarn Node Label 可以对每个节点进行标记，具有相同标签的节点为一个分区，提交任务至队列时，可以将容器运行在队列所关联的分区上。若不做任何配置，每个集群都具有一个 DEFAULT 分区，它包含了所有 NodeManager 所在的节点。

在 EMR 5.19.0 版本及更高版本中，EMR 已经自动修改了 yarn-site.xml 和 capacity-scheduler.xml 配置文件，为 Core 节点添加了“CORE”标签，并启用了节点标签功能。EMR 6.x 版本以上默认禁用该功能，需要手动启用，详见 Yarn 节点标签启用方法。

Yarn 节点分区分为两种类型：1）Exclusive：容器仅会被分配到与该分区标签匹配的节点上；2）Non-exclusive：若分区上有空闲资源可用，则该分区的资源会与 DEFAULT 分区或其他 Non-exclusive 分区共享资源。

在启用了节点标签功能的 EMR 集群中，Core 节点被配置为了Non-exclusive 的分区，提交任务后，AM 会默认只运行在具有“CORE”标签的节点上，保障了 AM 不会因为 Task 节点被回收而终止，防止了整个任务失败；但同时也会出现一种情况，由于 Core 节点为 Non-exclusive 分区，Executor 就有可能会运行在 Core 节点上，当大任务进来之后，Core 节点资源会被Executor占满，新提交任务的 AM 就会被pending，即使当前集群仍有闲置资源跑完一个小任务，但由于无法启动 AM，导致任务无法运行，对于大小任务混跑的集群很不友好。

因此，我们需要将 Core 节点所在分区配置为 Exclusive 分区，来保证在 Core 节点上仅运行 AM，避免资源共享，通过以下命令可实现该配置：

yarn rmadmin -removeFromClusterNodeLabels "CORE"
yarn cluster --list-node-labels
yarn rmadmin -addToClusterNodeLabels "CORE(exclusive=true)"

注意，这些命令需要在 EMR 集群 bootstrap 过程完成后执行。配置完成后，集群分区类型变化如下图所示：

在这时提交任务 Spark 任务后，Core 节点上仅存在 AM，Spot 实例的中断不会影响 Spark 任务进程，由于 AM 记录了任务的状态，当 EMR 补充到新的资源后，Spark 任务将会从中断处继续执行。

实验

我们将进行以下实验，来模拟 Spot 实例中断时，对 Spark 任务的影响。实验环境准备：

实验所在 AWS 区域：us-east-1；
创建 EMR 集群，以 emr-5.36.0 版本为例，包含 1 台 Master 节点，2 台 Core 节点和 1 台 Task 节点；
下载实验代码：

git clone https://github.com/micxyj/emr-spot-spark.git

修改 script.py，将代码最后一行中 “YOUR_BUCKET_NAME”替换成您的 S3 存储桶名；
在您的 S3 桶中分别创建两个路径，命名为 data 和 scripts，将下载的 food_establishment_data.csv 和 script.py 文件分别上传至刚刚创建的两个路径中。

实验步骤：

登陆到任意一台 EMR 节点，通过上一节中的命令，将 Core 节点设置为 Exclusive 类型分区，配置完成后，Yarn Node Label 界面如下图所示：

执行以下命令，提交 Spark 任务，该 script 逻辑为通过 sparksql 从原数据中查询结果，每隔固定时间，将结果写入新的路径下，我们会在第一次写入结果后停止 EMR Task 节点来模拟 Spot 节点中断，来观察 Spark 任务情况。

aws emr add-steps --cluster-id YOUR_CLUSTER_ID \
--steps Type=Spark,Name="Spark Program",ActionOnFailure=CONTINUE,\
Args=[--deploy-mode,cluster,--master,yarn,\
s3:// YOUR_BUCKET_NAME/scripts/script.py]

将“YOUR_CLUSTER_ID”和“YOUR_BUCKET_NAME”替换为您自己的集群 ID 和 S3 桶名称。

在第一次结果写入完成之后，在 EC2 控制台终止 Task 节点。

观察到集群正在 Resizing 状态，尝试补充新的节点：

在 Yarn 界面打开应用的 Tracking URL：

打开 Executor 界面，发现运行在 Core 节点上的 driver 还在 Active 状态，而运行在 Task 节点上的 Executor 为 Blacklisted 状态，由于 Core 分区类型为 Exclusive 类型，不共享资源，因此即使 Core 节点有充足资源，Executor 也没有运行在其上。