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

　　上一节我们完成了Flink单机模式的部署，本节讲解独立集群（standalonecluster）怎么部署。部署过程比较简单，只需要在单机模式的基础上指定JobManager地址、设置TaskManager列表以及设置ssh免密登录。本节第二部分讲解独立集群高可用配置。下面我们以三个节点为例，搭建Flink独立集群。
　　首先准备好搭建集群用的测试机，假定机器ip和主机名如下：192。168。1。10host1。test。com192。168。1。11host2。test。com192。168。1。12host3。test。com我们把host1。test。com作为JobManager，host2。test。com、host3。test。com作为TaskManager。1。3。1安装步骤
　　1。解压Flink软件包，切换工作目录到optflink1。14。3tarzxvfflink1。14。3binscala2。12。tgzCoptcdoptflink1。14。3
　　2。编辑文件confflinkconf。yaml，设置JobManager服务地址，用于JobManager和TaskManager交互调用。使用host1。test。com作为JobManager地址jobmanager。rpc。address：host1。test。com
　　3。编辑文件confworkers，指定TaskManager节点列表，每个节点占一行。host2。test。comhost3。test。com
　　4。设置ssh免密码登录。为了方便用户启停集群，当在主节点启动或者停止集群的时候，Flink需要ssh访问其他节点，然后一并执行启动或者停止命令，所以需要配置主节点到其他节点的ssh免密登录。如果不配置ssh免密，执行startcluster、stopcluster会提示用户输入机器密码。下面设置host1。test。com到host2。test。com、host3。test。com的免密码登录。
　　1）登录主机host1。test。com，执行sshkeygen命令回车生成公钥私钥对。参数rsa是加密算法，执行过程中会提示用户输入密码，我们直接回车不设置密码。sshkeygentrsa
　　2）复制公钥idrsa。pub到授权文件authorizedkeyscat。sshidrsa。pub。sshauthorizedkeys
　　3）修改授权文件authorizedkeys和。ssh目录权限chmod600。sshauthorizedkeyschmod700。ssh
　　4）复制授权文件authorizedkeys到host2。test。com、host3。test。com节点，如果目标节点不存在。ssh目录则需提前创建。如果目标节点authorizedkeys文件已存在且有内容，则应该追加idrsa。pub内容到文件末尾scp。sshauthorizedkeysroothost2。test。com：root。sshscp。sshauthorizedkeysroothost3。test。com：root。ssh
　　注意：所有节点的授权文件authorizedkeys权限必须是600，所有节点的。ssh目录的权限必须是700
　　5。复制配置好的Flink软件包到host2、host3机器，注意：Flink要求各节点安装路径必须保持一致复制到host2。test。comscproptflink1。14。3host2。test。com：opt复制到host3。test。comscproptflink1。14。3host3。test。com：opt
　　6。启动Flink独立集群在host1。test。com执行cdoptflink1。14。3启动集群（停止集群使用命令：。binstopcluster。sh）。binstartcluster。sh
　　从浏览器访问http：host1。test。com：8081页面，如果能看到如下图的Flinkweb管理页面，则证明Flink启动成功。页面中有显示TaskManager的数量是2。
　　1。3。2设置高可用
　　上面部署的独立集群有2个TaskManager，但只有一个JobManager，还需要解决JobManager的高可用问题。JobManager用于协调每个Flink任务的调度和资源管理。默认情况下，每个Flink集群只有一个JobManager实例，那么就意味着存在单点故障。如果JobManager崩溃，就不能提交新的任务，且运行中的任务也会失败。JobManager高可用可以在JobManager挂掉后，恢复JobManager，从而消除单点故障。
　　Flink提供了两种高可用方式实现：Zookeeper：基于ZooKeeper的高可用服务支持任意部署模式的Flink集群。它们需要外部的ZooKeeper集群。ApacheZookeeper提供分布式协调服务，解决一致性问题，被广泛应用于各大开源项目中，比如hadoop、hbase、kafka等。Kubernetes：基于Kubernetes的高可用服务只能用于FlinkonKubernetes集群。
　　独立集群高可用配置步骤：
　　1。在host1。test。com节点编辑配置文件：confflinkconf。yaml，设置高可用参数。必填参数highavailability：zookeeper必填参数，多quorum格式：host1：port1，host2：port2，host3：port3highavailability。zookeeper。quorum：host1。test。com：2181必填参数，storageDir需要是所有节点都能访问的地址，比如HDFS，S3，Ceph，nfs。如果使用hdfs地址则需要引入hadoop相关包，另外flink的启动用户必须有权限创建hdfs目录highavailability。storageDir：hdfs：flinkrecovery可选参数，存储状态的zookeeper路径highavailability。zookeeper。path。root：flink可选参数，当多个Flink集群使用同一个zookeeper地址时需要更改此参数highavailability。clusterid：default
　　注意：如果使用hdfs作为storageDir存储地址，需要引入hadoop相关类。修改etcprofile文件，在文件末尾增加行：前提是hadoop加入了环境变量，即是执行hadoopclasspath命令能打印classpath信息exportHADOOPCLASSPATHhadoopclasspath修改完成后执行sourceetcprofile刷新环境变量
　　2。编辑confmasters文件，添加JobManager，我们把host1。test。com、host2。test。com都作为JobManager角色。host1。test。com：8081host2。test。com：8081
　　3。编辑confworkers文件，添加TaskManager，我们把host2。test。com、host3。test。com都作为TaskManager角色。host2。test。comhost3。test。com
　　4。复制修改后的flinkconf。yaml、masters、workers文件到host2。test。com、host3。test。com节点scpconfflinkconf。yamlconfmastersconfworkershost2。test。com：optflink1。14。3confscpconfflinkconf。yamlconfmastersconfworkershost3。test。com：optflink1。14。3conf
　　5。启动zookeeper，执行。binstartzookeeperquorum。sh命令会在本机启动一个zookeeper服务，可通过confzoo。cfg修改zookeeper配置。用户也可以使用外部已部署好的zookeeper服务，确保在flinkconf。yaml文件的highavailability。zookeeper。quorum配置项值可访问。
　　6。启动Flink集群，。binstartcluster。sh同样启动成功后，在浏览器中访问某个JobManager可看到页面，访问另一个JobManager会提示：｛errors：〔Servicetemporarilyunavailableduetoanongoingleaderelection。Pleaserefresh。〕｝
　　启动JobManager需要在HDFS创建flink目录，如果没有权限会提示如下类似错误，用户去更改目录权限即可。Causedby：org。apache。hadoop。ipc。RemoteException（org。apache。hadoop。security。AccessControlException）：Permissiondenied：userroot，accessWRITE，inode：hdfs：supergroup：drwxrwxrxatorg。apache。hadoop。hdfs。server。namenode。FSPermissionChecker。check（FSPermissionChecker。java：504）atorg。apache。hadoop。hdfs。server。namenode。FSPermissionChecker。checkPermission（FSPermissionChecker。java：336）
　　如果有任何疑问欢迎留言，笔者头条号与公众号同名：小圈数据