flink高可用（standlone）的配置和问题解决

首先还是修改flink安装目录的conf目录下flink-conf.yaml文件，找到如下的三个配置，把原本的注释放开，然后配置自己的hdfs地址和zookeeper地址。

需要注意的是，我这里的hdfs是之前的ha集群，mycluster是我的hdfs的集群名，至于后边的内容会在hdfs中创建路径，可以自定义，不需要提前创建。

1 high-availability: zookeeper
2 high-availability.storageDir: hdfs://mycluster/flink/ha/
3 high-availability.zookeeper.quorum: node7-2:2181,node7-3:2181,node7-4:2181

workers修改

上一篇有提到过，旧版本的flink中有个文件叫slaves，新版的就叫这个workers，代表的是taskManger节点，之前我配置了三个，现在其中一个换成jobManager，所以删掉一个之后内容如下：

node7-2
node7-3
node7-4

masters修改

之前的监看flink集群搭建时，是没有管这个文件的，因为jobManager就只有一个，现在有了两个jobManager，就需要修改这个文件制定jobManager集群的节点。

实际上从这里，尤其是之前的masters和slaves这两个文件的命令，也很容易看出来他们的主从关系。

修改后的masters文件内容如下：

node7-1:8081
node7-2:8081

配置文件同步分发

和hdfs一样，和flink简单集群一样，这些修改的配置文件也都需要同步分发到所有的节点中，scp就不多说了。

hadoop依赖jar下载

上边操作完成后，我就使用start-cluster.sh启动的集群，然后看到打印出了如下的信息：

也没有报错，我以为就成功了，但是当我访问web页面时，无论是http://node001:8081还是http://node002:8081都无法访问，于是查看了flink的日志文件，结果发现日志中打印了如下的异常信息：

看起来就是无法识别和连接hdfs，实际上是因为没有相关的依赖，因此需要下载flink依赖的hadoop的jar到flink安装目录下的lib目录下。

这个插件在flink官网可以找到，https://flink.apache.org/downloads.html，这个连接中Additional Components下就是flink依赖的hadoop插件。

按网上说的，需要根据相应的hadoop版本下载对应的插件版本，但是我的hadoop是3.1.3，而这个页面中最高才是2.8.3，因此最终就下载了这个版本。

之后重新执行start-cluster.sh后日志没有再打印上边的异常，同时web页面也都可以成功打开了，并能看到两个taskManger。

在web页面提交上一次做好的flink程序的jar之后，也能看到running状态，似乎ha模式搭建成功了，但是实际上并不是。

classpath配置

当在上述jar生成的task运行的机器节点使用nc -lk 8888启动监听并发送数据后，web页面的Stdout上并没有如愿输出单词的统计次数，反而是在对应机器节点的日志中出现了异常。

通过查看，我的这个task运行在node003节点，找到日志使用tail -500 flink-root-taskexecutor-1-node003.log后发现了如下的异常：

StandaloneSessionClusterEntrypoint down with application status FAILED. Diagnostics java.io.IOException: Could not create FileSystem for highly available storage path (hdfs://jh/flink/ha/flinkCluster)
at org.apache.flink.runtime.blob.BlobUtils.createFileSystemBlobStore(BlobUtils.java:103)
at org.apache.flink.runtime.blob.BlobUtils.createBlobStoreFromConfig(BlobUtils.java:89)
at org.apache.flink.runtime.highavailability.HighAvailabilityServicesUtils.createHighAvailabilityServices(HighAvailabilityServicesUtils.java:117)
at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.createHaServices(ClusterEntrypoint.java:306)
at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.initializeServices(ClusterEntrypoint.java:269)
at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.runCluster(ClusterEntrypoint.java:211)
at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.lambda$startCluster$0(ClusterEntrypoint.java:172)
at org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:30)
at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.startCluster(ClusterEntrypoint.java:171)
at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.runClusterEntrypoint(ClusterEntrypoint.java:520)
at org.apache.flink.runtime.entrypoint.StandaloneSessionClusterEntrypoint.main(StandaloneSessionClusterEntrypoint.java:64)

经过一番查询和尝试后找到了解决办法，即配置两个环境变量，环境变量的配置方式较多，可以配系统变量，可以配用户变量，我就直接配置的系统变量，执行vi /etc/profile，然后加入如下两行：

export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop
export HADOOP_CLASSPATH=`hadoop classpath`

配置完成后使用source /etc/profile重新加载刚修改的内容，然后重新提交flink程序jar后日志不在报错，同时再次在nc中输入单词后，在web界面的Stdout中便能成功的刷新出预想的结果，至此，flink的ha模式搭建成功，搭建过程也算是对flink的设计思想和架构有了更进一步的认识。