部署高可用的EMR集群，为您的业务连续性保驾护航-InfoQ



 写点什么

Amazon EMR 是一个托管集群平台，可简化在 AWS 上运行大数据框架（如 Apache Hadoop 和 Apache Spark）以处理和分析海量数据的操作。借助这些框架和相关的开源项目 (如 Apache Hive 和 Apache Pig)。您可以处理用于分析目的的数据和商业智能工作负载。此外，您可以使用 Amazon EMR 转换大量数据和将大量数据移入和移出其他 AWS 数据存储和数据库，如 Amazon S3 或 Amazon DynamoDB 等。

以前，在 EMR 单主节点的情况下，会有单点失败的问题。当主节点出现故障时，整个集群会被终止，从而导致正在运行的任务执行失败。今天，我们很高兴地告诉您：EMR 在 5.23.0 的版本上推出了高可用（High Availability）的功能。您现在可以启动一个带有三个主节点的 EMR 集群，来实现多个应用的高可用，包括：YARN Resource Manager， HDFS Name Node，HBase，Hive，Spark 和 Ganglia 等。当 active 的主节点出现故障时，比如 Resource Manager 或 Name Node 崩溃，EMR 会自动切换到 standby 的主节点，新的主节点会保留和之前主节点相同的配置和引导操作（boot-strap actions）。

阅读本文，您将会了解到：

如何启用一个高可用（HA）的 EMR 集群
HDFS HA 的工作原理和配置变化
Yarn Resource Manager HA 的工作原理和配置变化
如何验证 EMR 的 HA
如何监控 EMR 的 HA

如何启用一个高可用的 EMR 集群

1.登陆到 EMR 的控制台，创建集群，点击“转到高级选项”

2.确保 EMR 的版本为 23.0，勾选需要的服务组件，勾选“启用多主控支持”

注：如果集群配置了 Hive，Hive HA 要求使用外部表元数据，所以建议在下面的“AWS Glue Data Catalog 设置”部分，勾选“用于 Hive 表元数据”

3.在硬件配置页，可以看到主节点数目已经是 3 个，选择合适的实例类型以及核心节点和任务节点数量。为了保证核心节点的高可用，官方建议是启动至少 4 个核心节点

4.点击下一步，按照向导配置对应参数，完成后点击“创建集群”。集群启动后，可以看到在集群的配置信息中已经有三个主节点的 DNS 域名列表

如何确认一个集群是否启用了 HA？

在 EMR 控制台中，选中要查看的集群，在“摘要”选项卡中的【网络和硬件】部分，如果主实例个数显示为 3，表面该集群启用了 HA

注意：

目前，我们还不能为一个正在运行的 EMR 集群启用 HA

HDFS HA 的工作原理和配置变化

【工作原理】

在一个高可用的 EMR 集群中， HDFS Name Node 运行在三个主节点中的其中两个。一个为 active 状态，另一个为 standby 状态。当 active 的 Name Node 出现故障时，standby 的 Name Node 会变为 active 并接管所有客户端的操作。EMR 会启用一个新的主节点将故障的 Name Node 替换，并设为 standby 状态。

您可以 SSH 登陆到任一台主节点上，运行如下命令来查看 active/standby 的 Name Node：

hdfs haadmin -getAllServiceState

输出示例如下：

ip-##-#-#-##1.us-west-2.compute.internal:8020     activeip-##-#-#-##2.us-west-2.compute.internal:8020    standby

复制代码

EMR HDFS 高可用的实现是使用 Quorum Journal Manager（QJM）。这里引入一个概念是 JournalNodes(JNs)，这是一组独立的 Daemon 来负责主备 Name Node 的状态信息同步。active 的 Name Node 会将 HDFS 文件系统的所有更新操作写入到 JournalNodes，standby 的 Name Node 会持续观察 JournalNodes 的编辑日志（edit log），并将这些操作实时地更新到自己的命名空间。同时，JournalNodes 还能保证同一时间只允许一个 Name Node 来写，避免“脑裂”导致的状态不一致和数据丢失。

【配置变化】

通过查看 HDFS 配置文件，我们可以看到在 hdfs-site.xml 配置文件中有如下新增或更新的属性：

ha.namenodes.ha-nn-uri （主备 Name Node 的 id）
namenode.rpc-address.ha-nn-uri.xxx（主备 Name Node 的 RPC 地址）
namenode.https-address.ha-nn-ur.xxx (主备 Name Node 的 HTTP 地址)
namenode.shared.edits.dir（JournalNodes 的地址组）
client.failover.proxy.provider.xxx（HDFS 客户端来获取 active Name Node 的 java 类）
ha.fencing.methods （在自动故障转移时用于隔离 active Name Node 的脚本或 java 类）
示例如下：

<property>
    <name>dfs.ha.namenodes.ha-nn-uri</name>
    <value>nn1,nn2</value>
  </property>
 
  <property>
    <name>dfs.namenode.rpc-address.ha-nn-uri.nn1</name>
    <value>ip-172-31-3-61.us-west-2.compute.internal:8020</value>
  </property>
 
  <property>
    <name>dfs.namenode.https-address.ha-nn-uri.nn1</name>
    <value>ip-172-31-3-61.us-west-2.compute.internal:50470</value>
  </property>
……
 
 <property>
    <name>dfs.namenode.shared.edits.dir</name>
    <value>qjournal://ip-172-31-3-61.us-west-2.compute.internal:8485;ip-172-31-7-112.us-west-2.compute.internal:8485;ip-172-31-10-171.us-west-2.compute.internal:8485/ha-nn-uri</value>
  </property>
 
  <property>
    <name>dfs.client.failover.proxy.provider.ha-nn-uri</name>
    <value>org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider</value>
  </property>
 
  <property>
    <name>dfs.ha.fencing.methods</name>
    <value>shell(/bin/true)</value>
  </property>

复制代码

另外，在 hdfs-site.xml 和 core-site.xml 配文件中，新增加了如下 2 个属性，表明启用了自动故障转移：

ha.automatic-failover.enabled
zookeeper.quorum
示例如下：

hdfs-site.xml：
<property>
    <name>dfs.ha.automatic-failover.enabled</name>
    <value>true</value>
  </property> 
core-site.xml：
 <property>
    <name>ha.zookeeper.quorum</name>
    <value>ip-172-31-3-61.us-west-2.compute.internal:2181,ip-172-31-7-112.us-west-2.compute.internal:2181,ip-172-31-10-171.us-west-2.compute.internal:2181</value>
  </property>

复制代码

关于 EMR HDFS 高可用实现的更多信息，您可以参考此链接。

Yarn Resource Manager HA 的工作原理和配置变化

###【工作原理】

与 HDFS 不同的是，Yarn ResourceManager(RM)运行在所有的三个主节点上。其中，一个 RM 是 active 状态，另外两个是 standby 状态。当 active 的 RM 出现故障时，EMR 会进行自动故障转移，将其中一个 standby 节点变为新的 active 来接管所有操作。之后 EMR 会将出现故障的 master 节点替换，并设为 standby 状态。

您可以 SSH 登陆到任一台主节点上，运行如下命令来查看当前的 active/standby RM:

yarn rmadmin -getAllServiceState

输出示例如下：

ip-##-#-#-##1.us-west-2.compute.internal:8033     standbyip-##-#-#-##2.us-west-2.compute.internal:8033    standbyip-##-#-#-##3.us-west-2.compute.internal:8033   active

复制代码

EMR RM 的 HA 使用了 RM Restart 的功能，在 RM 发生故障自动恢复后，新的 active RM 会从内部的状态存储中尽力读取之前 active RM 留下的状态信息，将未完成的应用继续运行。EMR 目前使用的是基于 zookeeper 的状态存储（ZooKeeper based state-store），该存储能够支持隔离机制（fencing mechanism），确保任何时间点只有一个 RM 去写状态存储，避免“脑裂”的发生。

###【配置变化】

与 HDFS HA 类似，通过查看 yarn 的配置文件，我们可以看到在 yarn-site.xml 配置文件中，有如下新增或更新的属性：

resourcemanager.ha.automatic-failover.enabled （启用 HA）
resourcemanager.ha.rm-ids （高可用的 RM id 列表）
resourcemanager.hostname.xx（RM hostname 列表）
resourcemanager.xxx.address.xx（RM web，管理等地址列表）
resourcemanager.recovery.enabled（启用 RM Recovery 的功能）
resourcemanager.store.class（指定 RM 使用基于 zookeeper 的状态存储）
resourcemanager.zk-address （ZK-quorum 的地址，用于状态存储和嵌入式领导选举）

示例如下：

<property>
    <name>yarn.resourcemanager.ha.automatic-failover.enabled</name>
    <value>true</value>
  </property>
 
  <property>
    <name>yarn.resourcemanager.ha.rm-ids</name>
    <value>rm1,rm2,rm3</value>
  </property>
…..
  <property>
    <name>yarn.resourcemanager.webapp.address.rm1</name>
    <value>ip-172-31-9-76.us-west-2.compute.internal:8088</value>
  </property>
 
  <property>
    <name>yarn.resourcemanager.hostname.rm1</name>
    <value>172.31.9.76</value>
  </property>
<property>
    <name>yarn.resourcemanager.recovery.enabled</name>
    <value>true</value>
  </property>
 
  <property>
    <name>yarn.resourcemanager.store.class</name>
    <value>org.apache.hadoop.yarn.server.resourcemanager.recovery.ZKRMStateStore</value>
  </property>
 
  <property>
    <name>yarn.resourcemanager.zk-address</name>
    <value>ip-172-31-9-76.us-west-2.compute.internal:2181,ip-172-31-3-219.us-west-2.compute.internal:2181,ip-172-31-4-251.us-west-2.compute.internal:2181</value>
  </property>

复制代码

关于 EMR RM HA 的更多信息，您可以参考此链接。

如何验证 EMR 的 HA

以 ResourceManager 为例，我们分别验证在没有任何 job 和有 job 运行时的自动故障转移。

没有 job 时 RM 故障转移：

1.查看当前 RM 的高可用状态，active 的 RM 是地址为 172.31.7.112 的实例

[hadoop@ip-172-31-7-112 hadoop-yarn]$ yarn rmadmin -getAllServiceState
ip-172-31-3-61.us-west-2.compute.internal:8033     standby
ip-172-31-7-112.us-west-2.compute.internal:8033    active
ip-172-31-10-171.us-west-2.compute.internal:8033   standby

复制代码

2.将 RM 进程停止，然后启动

[hadoop@ip-172-31-7-112 hadoop-yarn]$ sudo stop hadoop-yarn-resourcemanager
hadoop-yarn-resourcemanager stop/waiting
[hadoop@ip-172-31-7-112 hadoop-yarn]$ sudo start hadoop-yarn-resourcemanager
hadoop-yarn-resourcemanager start/running, process 5608

复制代码

3.再次查看 RM 状态，我们可以看到 active 的 RM 已经切换为 172.31.10.171 这台实例

[hadoop@ip-172-31-7-112 hadoop-yarn]$ yarn rmadmin -getAllServiceState
ip-172-31-3-61.us-west-2.compute.internal:8033     standby
ip-172-31-7-112.us-west-2.compute.internal:8033    standby
ip-172-31-10-171.us-west-2.compute.internal:8033   active

复制代码

在 job 运行时 RM 自动故障转移：

我们运行一个示例的 hive job，该 job 会创建一个基于 s3 的 hive 表，并将样例数据（CloudFront 日志）做聚合查询，统计出每种 OS 的请求数。您可以通过直接运行如下命令来提交示例的 hive job：

aws emr add-steps --cluster-id <your cluster id> --steps Type=HIVE,Name='Caculate number of request per OS',ActionOnFailure=CONTINUE,Args=[-f,s3://us-west-2.elasticmapreduce.samples/cloudfront/code/Hive_CloudFront.q,-d,INPUT=s3://us-west-2.elasticmapreduce.samples,-d,OUTPUT=s3://your/s3/output/path] --region us-west-2

复制代码

注意：请确保您具有运行 EMR step 相关的 IAM 权限，填写您的 EMR cluster id，s3://your/s3/output/path 替换为您 S3 的路径，作为 hive 查询结果的数据输出路径，region 与您 EMR cluster 所启用的 region 保持一致。

1.首先，查看当前 active 的 RM 是 172.31.10.171：

[hadoop@ip-172-31-10-171 ~]$ yarn rmadmin -getAllServiceState
ip-172-31-3-61.us-west-2.compute.internal:8033     standby
ip-172-31-7-112.us-west-2.compute.internal:8033    standby
ip-172-31-10-171.us-west-2.compute.internal:8033   active

复制代码

2.运行以上 hive job，当观察到 step 处于 running 状态后，通过 EC2 控制台或 AWS cli 将 172.31.10.171 这台实例终止（注意在终止前需要禁用终止保护：选择实例，点击操作 -》实例设置-》更改终止保护，点击请禁用）。

3.随后，可以观察到 EMR 自动启动一台新的实例作为 standby 的主节点。新的主节点仍然沿用之前节点的 private IP 地址，以确保各个节点间的正常通信，但 public IP 地址会发生改变。

4.Step 继续运行直到成功，查看 S3 对应路径下有如下预期的结果：

Android855
Linux813
MacOS852
OSX799
Windows883
iOS794

复制代码

在新的 active RM 上，我们可以看到在 RM 故障转移时有类似如下的日志，它首先检查是否有旧的 active RM 需要隔离，然后切换到 active 状态，并从 RMStateStore 中读取所有 application 的状态，将未完成的 application 继续运行：

2019-06-02 06:05:24,078 INFO org.apache.hadoop.ha.ActiveStandbyElector (main-EventThread): Checking for any old active which needs to be fenced...
2019-06-02 06:05:24,079 INFO org.apache.hadoop.ha.ActiveStandbyElector (main-EventThread): No old node to fence
2019-06-02 06:05:24,079 INFO org.apache.hadoop.ha.ActiveStandbyElector (main-EventThread): Writing znode /yarn-leader-election/ha-rm-uri/ActiveBreadCrumb to indicate that the local node is the most recent active...
…………..
2019-06-02 06:05:24,125 INFO org.apache.hadoop.yarn.server.resourcemanager.ResourceManager (main-EventThread): Transitioning to active state
2019-06-02 06:05:24,142 INFO org.apache.hadoop.yarn.server.resourcemanager.ResourceManager (main-EventThread): Recovery started
 
2019-06-02 06:05:24,142 INFO org.apache.hadoop.util.JvmPauseMonitor (org.apache.hadoop.util.JvmPauseMonitor$Monitor@47f6f422): Starting JVM pause monitor
2019-06-02 06:05:24,144 INFO org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore (main-EventThread): Loaded RM state version info 1.3
…
2019-06-02 06:05:25,482 INFO org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppImpl (main-EventThread): Recovering app: application_1559362602877_0014 with 1 attempts and final state = null

复制代码

需要注意的是，新的 active 的 RM 会尽力恢复之前 job 的状态信息，但仍有一定概率 job 会失败（比如中间数据丢失），这种情况建议您重新运行 job。

5.再次查看当前的 active 的 ResourceManager，可以看到 active 的已经切换到 172.31.3.61 这台实例。至此，高可用验证完成。

[hadoop@ip-172-31-10-171 ~]$ yarn rmadmin -getAllServiceState
ip-172-31-3-61.us-west-2.compute.internal:8033     active
ip-172-31-7-112.us-west-2.compute.internal:8033    standby
ip-172-31-10-171.us-west-2.compute.internal:8033   standby

复制代码