戳更多文章：

1-Flink入门

2-本地环境搭建&构建第一个Flink应用

3-DataSet API

4-DataSteam API

5-集群部署

6-分布式缓存

7-重启策略

8-Flink中的窗口

9-Flink中的Time

Flink时间戳和水印

Broadcast广播变量

FlinkTable&SQL

Flink实战项目实时热销排行

Flink写入RedisSink

17-Flink消费Kafka写入Mysql

Flink的HA搭建并不复杂，本质来说就是配置2个jobmanager。
本文作为Flink集群部署的补充篇。

这篇文章来自网络，向作者尼小摩致敬，

概述

JobManager 协调每个 Flink 部署。它负责调度和资源管理。
默认情况下，每个 Flink 集群只有一个 JobManager 实例。 这会产生单点故障(SPOF)：如果 JobManager 崩溃，则无法提交新作业并且导致运行中的作业运行失败。
使用 JobManager 高可用性模式，可以避免这个问题，从而消除 SPOF。您可以为Standalone和 YARN 集群配置高可用性。

Standalone集群高可用性

针对 Standalone 集群的 JobManager 高可用性的一般概念是，任何时候都有一个 主 JobManager 和多个备 JobManagers，以便在主节点失败时有备 JobManagers 来接管集群。这保证了没有单点故障，一旦备 JobManager 接管集群，作业就可以正常运行。主备 JobManager 实例之间没有明显的区别。每个 JobManager 都可以充当主备节点。
例如，请考虑以下三个 JobManager 实例的设置:

image.png

配置

要启用 JobManager 高可用性，您必须将高可用性模式设置为 zookeeper，配置 zookeeper quorum 将所有 JobManager 主机及其 web UI 端口写入配置文件。
Flink利用 ZooKeeper 在所有正在运行的 JobManager 实例之间进行分布式协调。 ZooKeeper 是独立于 Flink 的服务，通过 Leader 选举和轻量级一致状态存储提供高可靠的分布式协调。

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 集群

1. 在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
high-availability.storageDir:    hdfs:///flink/recovery

1. 在 conf/master 中配置 master:

    localhost:8081localhost:8082 

1. 在 conf/zoo.cfg 中配置 ZooKeeper 服务（目前，每台机器只能运行一个 ZooKeeper 进程）

    server.0=localhost:2888:3888

1. 启动 ZooKeeper quorum：

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

1. 启动 Flink HA 集群：

    $ bin/start-cluster.shStarting 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. 

1. 停止 Zookeeper quorum 和集群：

    $ bin/stop-cluster.shStopping 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 个数中。重要的是要注意 yarn.resourcemanager.am.max-attempts 为yarn中程序重启上限。因此， Flink 中设置的程序尝试次数不能超过 YARN 的集群设置。

示例：高可用的YARN Session

1.配置 HA 模式和 ZooKeeper 集群在 conf/flink-conf.yaml 中：

    high-availability: zookeeperhigh-availability.zookeeper.quorum: localhost:2181high-availability.storageDir: hdfs:///flink/recoveryhigh-availability.zookeeper.path.root: /flinkyarn.application-attempts: 10

1. 配置 ZooKeeper 服务在 conf/zoo.cfg 中(目前每台机器只能运行一个 ZooKeeper 进程)：

    server.0=localhost:2888:3888

1. 启动 ZooKeeper 集群：

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

1. 启动 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。