JobManager高可用性（HA）

作业管理器JobManager协调每个Flink部署组件，它负责调度以及资源管理。
默认情况下，每个Flink集群只有一个独立的JobManager实例，因此可能会产生单点故障（SPOF）。
使用JobManager High Availability，可以从JobManager的故障中恢复，从而消除SPOF。可以为独立（Standalone）集群和YARN集群配置高可用性。

独立（Standalone）集群高可用

独立集群的JobManager高可用性的一般概念是，任何时候都有一个独立拥有leader角色的JobManager和多个备用（Standby）的JobManagers，以便在leader失败时接管leader角色。这样就保证了没有单点故障，一旦备用JobManager取得领导角色，程序就可以正常运行。备用和主JobManager实例之间没有明确的区别，每个JobManager都可以充当主服务器或备用服务器。
例如，请考虑以下三个JobManager实例的设置：

Apache Flink 官方文档--作业管理器(JM, JobManager)高可用（HA）

配置

要启用JobManager高可用性，您必须将高可用性模式(high-availability mode)设置为zookeeper，配置ZooKeeper仲裁并设置包含所有JobManagers主机及其Web UI端口的master文件。
Flink利用ZooKeeper在所有正在运行的JobManager实例之间进行分布式协调。ZooKeeper是独立于Flink之外的服务，通过领导者选举和轻量级一致状态存储提供高度可靠的分布式协调服务。有关ZooKeeper的更多信息，请查看ZooKeeper的入门指南。Flink包含用于引导简单ZooKeeper安装的脚本。

masters文件:

要启动HA群集，请在以下位置配置主文件conf/masters：

masters文件： masters文件包括所有启动JobManager的主机节点，以及web用户界面绑定的端口号。
```
jobManagerAddress1:webUIPort1
[...]
jobManagerAddressX:webUIPortX
```
默认情况下，JobManager将为进程间通信（IPC）选择一个随机端口。可以通过
更改配置项high-availability.jobmanager.port 的值。此配置项接受单个端口（例如50010），范围（50000-50025）或两者的组合（50010,50011,50020-50025,50050-50075）。

flink-conf.yaml文件：

要启动HA群集，请将以下配置项添加到conf/flink-conf.yaml：

high-availability mode (required)：high-availability mode必须在conf/fink-conf.yaml文件中设置为zookeeper，以启用高可用模式。或者该选项可以设置为Flink用来创建高可用服务实例的工厂类或者FQN。
```
high-availability: zookeeper
```
ZooKeeper quorum (required): ZooKeeper仲裁是ZooKeeper服务的复制组，它提供分布式协调服务。
```
high-availability.zookeeper.quorum: address1:2181[,...],addressX:2181
```
每个addressX：port指的是一个ZooKeeper服务，Flink可以通过给定的地址和端口访问它。
ZooKeeper root (recommended): ZooKeeper根节点，在该节点下放置所有集群节点。
```
high-availability.zookeeper.path.root: /flink
```

ZooKeeper cluster-id (recommended): ZooKeeper cluster-id 节点，在该节点下放置集群所需的所有协调数据。


high-availability.cluster-id: /default_ns  # important: customize per cluster
```集
> **重要**: 每个作业以一个YARN会话的形式在YARN集群运行，或其他集群管理器运行时，不应手动设置此值。在这些情况下，将会根据应用程序ID自动生成cluster-id。手动设置cluster-id会覆盖YARN中的此行为。反过来，使用“-z CLI”选项指定cluster-id会覆盖此手动配置项。如果在裸机上运行多个Flink HA集群，则必须为每个群集手动配置单独的集群ID。

Storage directory (required): JobManager元数据保存在文件系统的storageDir中，而指向此状态的指针存储在ZooKeeper中。
```
high-availability.storageDir: hdfs:///flink/recovery
```
storageDir存储所有元数据，恢复JobManager失败需要的元数据。

配置主服务器和ZooKeeper仲裁后，可以像往常一样使用提供的集群启动脚本。他们将启动HA群集。请记住，调用脚本时ZooKeeper仲裁必须运行，并确保要为启动的每个HA群集配置单独的ZooKeeper根路径。

示例：拥有2个JobManager的独立集群

在文件conf/flink-conf.yaml中配置高可用模式以及zookeeper仲裁集群

high-availability: zookeeper
high-availability.zookeeper.quorum: localhost:2181
high-availability.zookeeper.path.root: /flink
high-availability.cluster-id: /cluster_one # important: customize per cluster
high-availability.storageDir: hdfs:///flink/recovery

在文件conf/masters中配置masters
```
localhost:8081
localhost:8082
```
在文件conf/zoo.cfg中配置Zookeeper服务器(目前只能是在每个机器节点上运行单一的Zookeeper服务)
```
server.0=localhost:2888:3888
```

启动Zookeeper仲裁

$ bin/start-zookeeper-quorum.sh
Starting zookeeper daemon on host localhost.

启动HA集群

$ bin/start-cluster.sh
Starting HA cluster with 2 masters and 1 peers in ZooKeeper quorum.
Starting jobmanager daemon on host localhost.
Starting jobmanager daemon on host localhost.
Starting taskmanager daemon on host localhost.

停止ZooKeeper仲裁和集群：

$ bin/stop-cluster.sh
Stopping taskmanager daemon (pid: 7647) on localhost.
Stopping jobmanager daemon (pid: 7495) on host localhost.
Stopping jobmanager daemon (pid: 7349) on host localhost.
$ bin/stop-zookeeper-quorum.sh
Stopping zookeeper daemon (pid: 7101) on host localhost.

YARN集群高可用

运行高可用YARN集群时，不会运行多个JobManager（ApplicationMaster）实例，而只运行一个，这些实例在发生故障时由YARN重新启动。确切的行为取决于使用YARN的特定版本。

配置

Application Master最大重试次数(yarn-site.xml)

必须配置YARN集群的yarn-site.xml文件中的Application Master最大重试次数。

<property>
  <name>yarn.resourcemanager.am.max-attempts</name>
  <value>4</value>
  <description>
    The maximum number of application master execution attempts.
  </description>
</property>

当前YARN版本的默认值为2，意味着可以容忍一次JobManager失败。

Application Attempts (flink-conf.yaml)

除了前述的HA配置项，还必须在fink-conf.yaml文件中配置最大重试次数。

yarn.application-attempts: 10

这意味着在YARN使应用程序失败之前，应用程序可以重新启动9次以进行失败尝试（9次重试+ 1次初始尝试）。如果YARN操作（抢占，节点硬件故障或重新启动，或NodeManager重新同步）需要，YARN可以执行其他重新启动。这些重启不计入yarn.application-attempts。重要的是要注意yarn.resourcemanager.am.max-attempts应用程序重新启动的上限。因此，Flink中设置的应用程序尝试次数不能超过启动的YARN集群设置。

容器关闭行为

YARN 2.3.0 < version < 2.4.0.如果Application Master失败，则重启所有的容器。
YARN 2.4.0 < version < 2.6.0.TaskManager容器在Application Master故障期间保持活动状态。具有以下优点：启动时间更快并且用户不必等待再次获得容器资源。
YARN 2.6.0 <= version.将尝试失败有效性间隔设置为Flinks的Akka超时值。尝试失败有效性间隔表示系统在一个间隔期间看到最大应用程序尝试次数后才会终止应用程序。这避免了持久的作业会耗尽它的应用程序尝试。
注意：Hadoop YARN 2.4.0有一个主要错误（在2.5.0中修复），阻止重新启动的Application Master / Job Manager容器重启容器。有关详细信息，请参阅FLINK-4142。我们建议至少在YARN上使用Hadoop 2.5.0进行高可用性设置。
示例：高可用的YARN会话
1. 在文件conf/flink-conf.yaml中配置高可用模式以及zookeeper仲裁集群
```
high-availability: zookeeper
high-availability.zookeeper.quorum: localhost:2181
high-availability.zookeeper.path.root: /flink
high-availability.storageDir: hdfs:///flink/recovery
yarn.application-attempts: 10
```
2. 在文件conf/zoo.cfg中配置Zookeeper服务器(目前只能是在每个机器节点上运行单一的Zookeeper服务)
```
server.0=localhost:2888:3888
```
3. 启动Zookeeper仲裁
```
$ bin/start-zookeeper-quorum.sh
Starting zookeeper daemon on host localhost.
```
  4.启动HA集群
```
$ bin/yarn-session.sh -n 2
```

Zookeeper启用安全的配置

如果ZooKeeper使用Kerberos以安全模式运行，则可以根据需要覆盖flink-conf.yaml中的以下配置：

# default is "zookeeper". If the ZooKeeper quorum is configured
# with a different service name then it can be supplied here.
zookeeper.sasl.service-name: zookeeper

# default is "Client". The value needs to match one of the values
# configured in "security.kerberos.login.contexts".
zookeeper.sasl.login-context-name: Client

有关Kerberos安全性的Flink配置的更多信息，请参阅此处。您还可以在此处找到有关Flink内部如何设置基于Kerberos的安全性的更多详细信息。

引导Zookeeper

如果没有正在运行的ZooKeeper安装，则可以使用Flink附带的帮助程序脚本。
有一个ZooKeeper配置模板conf/zoo.cfg。可以将主机配置为使用server.X培香项运行ZooKeeper ，其中X是每个服务器的唯一ID：

server.X = addressX：peerPort：leaderPort
[...]
server.Y = addressY：peerPort：leaderPort

脚本bin/start-zookeeper-quorum.sh将在每个配置的主机上启动ZooKeeper服务器。启动的进程通过Flink包装器启动ZooKeeper服务器，该包装器从conf/zoo.cfg中读取配置，并确保为方便起见设置一些必需的配置值。在生产环境中，建议部署管理独立的ZooKeeper集群。

Apache Flink 官方文档--作业管理器(JM, JobManager)高可用HA