1。3部署Flink独立集群Flink核心源码解读

　　上一节我们完成了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)
　　如果有任何疑问欢迎留言，笔者头条号与公众号同名：小圈数据