上一节我们完成了Flink单机模式的部署，本节讲解独立集群（standalone cluster）怎么部署。部署过程比较简单，只需要在单机模式的基础上指定JobManager地址、设置TaskManager列表以及设置ssh免密登录。本节第二部分讲解独立集群高可用配置。下面我们以三个节点为例，搭建Flink独立集群。

首先准备好搭建集群用的测试机，假定机器ip和主机名如下： 192.168.1.10 host1.test.com 192.168.1.11 host2.test.com 192.168.1.12 host3.test.com 我们把host1.test.com作为JobManager， host2.test.com、host3.test.com作为TaskManager。

1.3.1 安装步骤

1.解压Flink软件包，切换工作目录到/opt/flink-1.14.3

tar zxvf flink-1.14.3-bin-scala_2.12.tgz -C /opt/
cd /opt/flink-1.14.3

2.编辑文件conf/flink-conf.yaml，设置JobManager服务地址，用于JobManager和TaskManager交互调用。

# 使用host1.test.com作为JobManager地址
jobmanager.rpc.address: host1.test.com

3.编辑文件conf/workers，指定TaskManager节点列表，每个节点占一行。

host2.test.com  
host3.test.com

4.设置ssh免密码登录。为了方便用户启停集群，当在主节点启动或者停止集群的时候，Flink需要ssh访问其他节点，然后一并执行启动或者停止命令，所以需要配置主节点到其他节点的ssh免密登录。如果不配置ssh免密，执行start-cluster、stop-cluster会提示用户输入机器密码。下面设置host1.test.com到host2.test.com、host3.test.com的免密码登录。

1）登录主机host1.test.com，执行ssh-keygen命令回车生成公钥私钥对。参数rsa是加密算法，执行过程中会提示用户输入密码，我们直接回车不设置密码。

ssh-keygen -t rsa

2）复制公钥id_rsa.pub到授权文件authorized_keys

cat ~/.ssh/id_rsa.pub  > ~/.ssh/authorized_keys

3）修改授权文件authorized_keys和.ssh目录权限

chmod 600 ~/.ssh/authorized_keys
chmod 700 ~/.ssh

4）复制授权文件authorized_keys到host2.test.com、host3.test.com节点，如果目标节点不存在.ssh目录则需提前创建。

#如果目标节点authorized_keys文件已存在且有内容，则应该追加id_rsa.pub内容到文件末尾
scp ~/.ssh/authorized_keys root@host2.test.com:/root/.ssh/
scp ~/.ssh/authorized_keys root@host3.test.com:/root/.ssh/

注意：所有节点的授权文件authorized_keys权限必须是600，所有节点的.ssh目录的权限必须是700

5.复制配置好的Flink软件包到host2、host3机器，注意：Flink要求各节点安装路径必须保持一致

#复制到host2.test.com
scp -r /opt/flink-1.14.3 host2.test.com:/opt
#复制到host3.test.com
scp -r /opt/flink-1.14.3 host3.test.com:/opt

6.启动Flink独立集群

#在host1.test.com执行
cd /opt/flink-1.14.3
#启动集群（停止集群使用命令：./bin/stop-cluster.sh）
./bin/start-cluster.sh

从浏览器访问http://host1.test.com:8081页面，如果能看到如下图的Flink web管理页面，则证明Flink启动成功。页面中有显示TaskManager的数量是2。

1.3.2 设置高可用

上面部署的独立集群有2个TaskManager，但只有一个JobManager，还需要解决JobManager的高可用问题。JobManager用于协调每个Flink任务的调度和资源管理。默认情况下，每个Flink集群只有一个JobManager实例，那么就意味着存在单点故障。如果JobManager崩溃，就不能提交新的任务，且运行中的任务也会失败。 JobManager高可用可以在JobManager挂掉后，恢复JobManager，从而消除单点故障。

Flink提供了两种高可用方式实现：

• Zookeeper：基于ZooKeeper的高可用服务支持任意部署模式的Flink集群。它们需要外部的ZooKeeper集群。Apache Zookeeper提供分布式协调服务，解决一致性问题，被广泛应用于各大开源项目中，比如hadoop、hbase、kafka等。
• Kubernetes：基于Kubernetes的高可用服务只能用于Flink on Kubernetes集群。

独立集群高可用配置步骤：

1.在host1.test.com节点编辑配置文件：conf/flink-conf.yaml，设置高可用参数。

#必填参数
high-availability: zookeeper   
#必填参数，多quorum格式：host1:port1,host2:port2,host3:port3
high-availability.zookeeper.quorum: host1.test.com:2181
#必填参数，storageDir需要是所有节点都能访问的地址，比如HDFS, S3, Ceph, nfs。如果使用hdfs地址则需要引入hadoop相关包，另外flink的启动用户必须有权限创建hdfs目录
high-availability.storageDir: hdfs:///flink/recovery

#可选参数，存储状态的zookeeper路径
high-availability.zookeeper.path.root: /flink
#可选参数，当多个Flink集群使用同一个zookeeper地址时需要更改此参数
high-availability.cluster-id: /default

注意： 如果使用hdfs作为storageDir存储地址，需要引入hadoop相关类。修改/etc/profile文件，在文件末尾增加行：

#前提是hadoop加入了环境变量，即是执行hadoop classpath命令能打印classpath信息
export HADOOP_CLASSPATH=`hadoop classpath`
#修改完成后执行source /etc/profile刷新环境变量

2.编辑conf/masters文件，添加JobManager，我们把host1.test.com、host2.test.com都作为JobManager角色。

host1.test.com:8081
host2.test.com:8081

3.编辑conf/workers文件，添加TaskManager，我们把host2.test.com、host3.test.com都作为TaskManager角色。

host2.test.com
host3.test.com

4.复制修改后的flink-conf.yaml、masters、workers文件到host2.test.com、host3.test.com节点

scp conf/flink-conf.yaml conf/masters conf/workers host2.test.com:/opt/flink-1.14.3/conf/
scp conf/flink-conf.yaml conf/masters conf/workers host3.test.com:/opt/flink-1.14.3/conf/

5.启动zookeeper，执行./bin/start-zookeeper-quorum.sh 命令会在本机启动一个zookeeper服务，可通过conf/zoo.cfg修改zookeeper配置。用户也可以使用外部已部署好的zookeeper服务，确保在flink-conf.yaml文件的high-availability.zookeeper.quorum配置项值可访问。

6.启动Flink集群，./bin/start-cluster.sh 同样启动成功后，在浏览器中访问某个JobManager可看到页面，访问另一个JobManager会提示：{"errors":["Service temporarily unavailable due to an ongoing leader election. Please refresh."]}

启动JobManager需要在HDFS创建/flink目录，如果没有权限会提示如下类似错误，用户去更改目录权限即可。

Caused by: org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.security.AccessControlException): Permission denied: user=root, access=WRITE, inode="/":hdfs:supergroup:drwxrwxr-x
at org.apache.hadoop.hdfs.server.namenode.FSPermissionChecker.check(FSPermissionChecker.java:504)
at org.apache.hadoop.hdfs.server.namenode.FSPermissionChecker.checkPermission(FSPermissionChecker.java:336)

如果有任何疑问欢迎留言，笔者头条号与公众号同名：小圈数据