概述
JobManager 协调每个 Flink 部署。它负责调度和资源管理。
默认情况下，每个 Flink 集群只有一个 JobManager 实例。 这会产生单点故障(SPOF)：如果 JobManager 崩溃，则无法提交新作业并且导致运行中的作业运行失败。
使用 JobManager 高可用性模式，可以避免这个问题，从而消除 SPOF。您可以为Standalone和 YARN 集群配置高可用性。
Standalone集群高可用性
针对 Standalone 集群的 JobManager 高可用性的一般概念是，任何时候都有一个 主 JobManager 和多个备 JobManagers，以便在主节点失败时有备 JobManagers 来接管集群。这保证了没有单点故障，一旦备 JobManager 接管集群，作业就可以正常运行。主备 JobManager 实例之间没有明显的区别。每个 JobManager 都可以充当主备节点。
例如，请考虑以下三个 JobManager 实例的设置:
image
配置
要启用 JobManager 高可用性，您必须将高可用性模式设置为 zookeeper，配置 zookeeper quorum 将所有 JobManager 主机及其 web UI 端口写入配置文件。
Flink利用 ZooKeeper 在所有正在运行的 JobManager 实例之间进行分布式协调。 ZooKeeper 是独立于 Flink 的服务，通过 Leader 选举和轻量级一致状态存储提供高可靠的分布式协调。 更多关于 ZooKeeper 的信息, 请查看 ZooKeeper 的入门指南 。 Flink 包含用于 Bootstrap ZooKeeper 安装的脚本。
Masters文件 (masters)
要启动HA集群，请在以下位置配置Master文件 conf/masters:
- masters文件：masters文件包含启动 jobmanager 的所有主机和 web 用户界面绑定的端口。
jobManagerAddress1:webUIPort1
[...]
jobManagerAddressX:webUIPortX
默认情况下，job manager选一个随机端口作为进程随机通信端口。您可以通过 high-availability.jobmanager.port 键修改此设置。此配置接受单个端口（例如50010），范围（50000-50025）或两者的组合（50010,50011,50020-50025,50050-50075）。
配置文件（flink-conf.yaml）
要启动HA集群，请将以下配置键添加到 conf/flink-conf.yaml:
- 高可用性模式（必需）：在 conf/flink-conf.yaml 中，必须将高可用性模式设置为zookeeper，以打开高可用模式。或者将此选项设置为工厂类的 FQN，Flink 通过创建 HighAvailabilityServices 实例使用。
high-availability: zookeeper
- Zookeeper quorum（必需）： ZooKeeper quorum 是 ZooKeeper 服务器的复制组，它提供分布式协调服务。
high-availability.zookeeper.quorum: address1:2181[,...],addressX:2181
每个 addressX:port 都是一个 ZooKeeper 服务器的ip及其端口，Flink 可以在指定的地址和端口访问zookeeper。
- ZooKeeper root （推荐）： ZooKeeper 根节点，在该节点下放置所有集群节点。
high-availability.zookeeper.path.root: /flink
- ZooKeeper cluster-id（推荐）： ZooKeeper的cluster-id节点，在该节点下放置集群的所有相关数据。
high-availability.cluster-id: /default_ns # important: customize per cluster
重要：在运行 YARN 或其他群集管理器中运行时，不要手动设置此值。在这些情况下，将根据应用程序 ID 自动生成 cluster-id。 手动设置 cluster-id 会覆盖 YARN 中的自动生成的 ID。反过来，使用 -z CLI 选项指定 cluster-id 会覆盖手动配置。如果在裸机上运行多个 Flink HA 集群，则必须为每个集群手动配置单独的 cluster-id。
- 存储目录（必需）： JobManager 元数据保存在文件系统 storageDir 中，在 ZooKeeper 中仅保存了指向此状态的指针。
high-availability.storageDir: hdfs:///flink/recovery
该storageDir 中保存了 JobManager 恢复状态所需的所有元数据。
配置 master 文件和 ZooKeeper quorum 之后，您可以使用提供的集群启动脚本。它们将启动 HA 群集。请注意，启动 Flink HA 集群前，必须启动 Zookeeper 集群，并确保为要启动的每个 HA 群集配置单独的 ZooKeeper 根路径。
示例：具有2个 JobManager 的 Standalone 集群
- 在conf/flink-conf.yaml 中配置高可用模式和 ZooKeeper quorum：
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/master 中配置 master:
localhost:8081
localhost:8082
- 在 conf/zoo.cfg 中配置 ZooKeeper 服务（目前，每台机器只能运行一个 ZooKeeper 进程）
server.0=localhost:2888:3888
- 启动 ZooKeeper quorum：
$ bin/start-zookeeper-quorum.sh
Starting zookeeper daemon on host localhost.
- 启动 Flink 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 quorum 和集群：
$ 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) 实例，而只运行一个，该JobManager实例失败时，YARN会将其重新启动。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配置(参考上文)之外，您还必须配置最大重试次数 conf/flink-conf.yaml:
yarn.application-attempts: 10
这意味着在如果程序启动失败，YARN会再重试9次（9 次重试 + 1次启动）。如果 YARN 操作需要，如果启动10次作业还失败，yarn才会将该任务的状态置为失败。如果抢占，节点硬件故障或重启，NodeManager 重新同步等操作需要，YARN继续尝试启动应用。 这些重启不计入 yarn.application-attempts 个数中，请参阅 Jian Fang 的博客文章 。重要的是要注意 yarn.resourcemanager.am.max-attempts 为yarn中程序重启上限。因此， Flink 中设置的程序尝试次数不能超过 YARN 的集群设置。
容器关闭行为
- YARN 2.3.0 < 版本 < 2.4.0。 如果application master进程失败，则所有的container都会重启。
- YARN 2.4.0 < 版本 < 2.6.0。 TTaskManager container在 application master 故障期间，会继续工作。这具有以下优点：启作业恢复时间更快，且缩短所有task manager启动时，申请资源的时间。
- YARN 2.6.0 <= version：将尝试失败有效性间隔设置为 Flink 的 Akka 超时值。尝试失败有效性间隔表示只有在系统在一个间隔期间看到最大应用程序尝试次数后才会终止应用程序。这避免了持久的工作会耗尽它的应用程序尝试。
注意: Hadoop YARN 2.4.0 有一个主要的 bug (在2.5.0中修复)，阻止重新启动的Application Master / Job Manager 重启容器。有关详细信息，请参阅 FLINK-4142 。我们建议，在yarn版本要等于或高于Hadoop 2.5.0 增加高可用配置。
示例：高可用的YARN Session
- 配置 HA 模式和 ZooKeeper 集群在 conf/flink-conf.yaml 中 ：
high-availability: zookeeper
high-availability.zookeeper.quorum: localhost:2181
high-availability.storageDir: hdfs:///flink/recovery
high-availability.zookeeper.path.root: /flink
yarn.application-attempts: 10
- 配置 ZooKeeper 服务在 conf/zoo.cfg 中(目前每台机器只能运行一个 ZooKeeper 进程)：
server.0=localhost:2888:3888
- 启动 ZooKeeper 集群：
$ bin/start-zookeeper-quorum.sh
Starting zookeeper daemon on host localhost.
- 启动 HA 集群：
$ bin / yarn-session.sh -n 2
配置 Zookeeper 安全性
如果 ZooKeeper 使用 Kerberos 以安全模式运行，flink-conf.yaml 根据需要覆盖以下配置：
zookeeper.sasl.service-name: zookeeper # 默认设置是 “zookeeper” 。如果 ZooKeeper 集群配置了
# 不同的服务名称，那么可以在这里提供。
zookeeper.sasl.login-context-name: Client # 默认设置是 “Client”。该值配置需要匹配
# "security.kerberos.login.contexts"中的其中一个值。
有关 Kerberos 安全性的 Flink 配置的更多信息，请参阅 此处。您还可以在 此处 找到关于 Flink 内部如何设置基于 kerberos 的安全性的详细信息。
Bootstrap ZooKeeper
如果您没有正在运行的ZooKeeper，则可以使用Flink程序附带的脚本。
这是一个 ZooKeeper 配置模板 conf/zoo.cfg。您可以为主机配置为使用 server.X 条目运行 ZooKeeper，其中 X 是每个服务器的唯一IP:
server.X=addressX:peerPort:leaderPort
[...]
server.Y=addressY:peerPort:leaderPort
该脚本 bin/start-zookeeper-quorum.sh 将在每个配置的主机上启动 ZooKeeper 服务器。 Flink wrapper 会启动 ZooKeeper 服务，该 wraper 从 conf/zoo.cfg 中读取配置，并设置一些必需的配置项。在生产设置中，建议您使用自己安装的 ZooKeeper。
