Flink的高可用集群环境

Flink的高可用集群环境

Flink简介

Flink核心是一个流式的数据流执行引擎,其针对数据流的分布式计算提供了数据分布,数据通信以及容错机制等功能。

因现在主要Flink这一块做先关方面的学习,因此准备要开通Apache Flink专栏这一块定期发布一些文章。今天在自己的博客因为专栏无法申请通过,所以先在此记录第一篇关于Flink部署的文章。

在这里顺便打个小广告,Flink社区第一季线下meetup,已在上海,北京举办。接下来分别会在成都和深圳举办接下来的几期,也希望小伙伴们踊跃的加入到Flink社区来,下载钉钉,扫描下方二维码即可加入大群。

首先今天先介绍一下Flink的安装,安装部署最新1.6版本支持有8种安装方式,详细可以参考安装部署方式【Clusters & Deployment】 。下面主要介绍Standalone Cluster模式和on yarn模式 。

软件包下载地址

一.Flink独立集群模式安装(Cluster Standalone)

1.1.解压安装

[root@h001 soft]# tar -zxvf flink-1.2.0-bin-hadoop26-scala_2.11.tgz -C /usr/bigdata/
1
1.2.Flink配置(Configuring Flink)

对其进行相关的配置。主要涉及到的配置文件是conf/flink-conf.yaml

flink-conf.yaml配置

jobmanager.rpc.address:值设置成你master节点的IP地址
taskmanager.heap.mb:每个TaskManager可用的总内存
taskmanager.numberOfTaskSlots:每台机器上可用CPU的总数
parallelism.default:每个Job运行时默认的并行度(这个参数在文档中介绍好像有问题)
taskmanager.tmp.dirs:临时目录
jobmanager.heap.mb:每个节点的JVM能够分配的最大内存
jobmanager.rpc.port: 6123
jobmanager.web.port: 8081

[root@h001 conf]# vim flink-conf.yaml
jobmanager.rpc.address:h001
taskmanager.heap.mb:2048
taskmanager.numberOfTaskSlots:4
parallelism.default:10
taskmanager.tmp.dirs:/tmp
jobmanager.heap.mb:2048
jobmanager.web.port: 8081
jobmanager.rpc.port: 6123
1
2
3
4
5
6
7
8
9
10
主节点与从节点配置

[root@h002 conf]# vim slaves
h002
h003
h004
h005
[root@h001 conf]# vim masters
h001:8082
1
2
3
4
5
6
7
1.3.Flink安装包分发到所有的worker节点上

[root@h001 bigdata]# clush -v -w h[002-005] --copy flink-1.5.1 --dest /usr/bigdata/
1
1.4.启动Flink(Starting Flink)

在master节点上运行下面的脚本,那么这台机器上将会启动一个JobManager,并通过SSH连接列在slaves文件中的所有节点以便在每个节点上启动TaskManager

[root@h001 flink-1.5.1]# bin/start-cluster.sh
1
如果停止集群,可以在master节点上运行下面的命令

[root@h001 flink-1.5.1]# bin/stop-cluster.sh
1
1.5. 在已经运行的集群中添加JobManager/TaskManager

通过bin/taskmanager.sh或者bin/jobmanager.sh脚本在已经运行的集群中添加JobManager或者TaskManager节点

[root@h001 flink-1.2.0]# bin/jobmanager.sh (start (local|cluster) [host] [webui-port]|stop|stop-all)

[root@h001 flink-1.2.0]# bin/taskmanager.sh (start (local|cluster) [host] [webui-port]|stop|stop-all)
1
2
3
二.JobManager高可用(HA)

JobManager协调每一个Flink集群环境,它负责作业调度和资源管理。默认情况下,一个Flink集群中只有一个JobManager实例,这很容易造成单点故障(SPOF)。如果JobManager奔溃了,那么将没有新的程序被提交,同时运行的程序将失败。
       对于JobManager高可用来说,我们可以从失败的JobManager中恢复,因此可以消除单点故障的问题。我们可以配置Standalone模式和YARN集群模式下的高可用JobManager的HA,是通过Zookeeper实现的,因此需要先搭建好Zookeeper集群,同时HA的信息,还要存储在HDFS中,因此也需要Hadoop集群,最后修改Flink中的配置文件。

根据部署方式不同,Flink Jobmanager HA配置分为2种:
        1、standalone cluster HA
        2、Yarn cluster HA

2.1. Standalone集群模式高可用

对于Standalone集群模式下的JobManager高可用通常的方案是:Flink集群的任一时刻只有一个leading JobManager,并且有多个standby JobManager。当leader失败后,standby通过选举出一个JobManager作为新的leader。这个方案可以保证没有单点故障的问题。对于standby和master JobManager实例来说,其实没有明确的区别,每一个JobManager能够当担master或standby角色。

2.1.1.相关配置

为了保证JobManager高可用,你需要设置Zookeeper为recovery mode(恢复模式),配置一个Zookeeper quorum并且对所有的JobManager节点和它们的Web UI端口号设置一个masters文件。

Flink引入Zookeeper的目的主要是让JobManager实现高可用(leader选举)
Flink使用Zookeeper在所有运行的JobManager实例中进行分布式调度的协调。Zookeeper在Flink中是一个独立的服务,它能够通过leader选举和轻量级的一致性状态存储来提供高度可靠的分布式协调器
Master File(masters)
为了启动一个HA-cluster,需要在conf/masters中配置masters。
masters文件:masters文件包含所有的hosts,每个host启动都JobManager,并且指定绑定的Web UI端口号:
jobManagerAddress1:webUIPort1
[...]
jobManagerAddressX:webUIPortX
1
2
3
配置文件flink-conf.yaml

为了启动一个HA-Cluster,需要在conf/flink-conf.yaml添加如下配置参数:

Recovery mode(必须的):recovery.mode: zookeeper
zookeeper quorum(必须的):recovery.zookeeper.quorum: address1:2181,...
Zookeeper root(推荐的):Flink在Zookeeper中的root节点,下面放置所有需要协调的数据recovery.zookeeper.path.root: /flink
1
2
3
      如果你运行多个Flink HA集群,那么你必须手工配置每个Flink集群使用独立的root节点

State backend and storage directory(必须的):JobManager元数据在statebackend保持并且仅仅在Zookeeper中存储,目前在HA模式中,仅支持filesystem。
state.backend: filesystem
state.backend.fs.checkpointdir:hdfs://namenode-host:port/flink-checkpoints
recovery.zookeeper.storageDir: hdfs:///recovery
recovery.zookeeper.storageDir指定的路径中存储了所有的元数据,用来恢复失败的JobManager
1
2
3
4
5
2.2. 两个JobManager的Standalone模式下的集群

conf/flink-conf.yaml文件

配置恢复模式和Zookeeper quorum

[root@h001 conf]# vim flink-conf.yaml
recovery.mode: zookeeper
recovery.zookeeper.quorum: h002:2181,h003:2181,h004:2181
recovery.zookeeper.path.root: /flink
state.backend: filesystem
state.backend.fs.checkpointdir: hdfs://h001:9000/flink/checkpoints
recovery.zookeeper.storageDir: hdfs://h001:9000/flink/recovery
1
2
3
4
5
6
7
在Hadoop文件系统创建文件夹

[root@h001 conf]# hadoop fs -mkdir -p /flink/checkpoints
[root@h001 conf]# hadoop fs -mkdir -p /flink/recovery
[root@h001 conf]# hadoop fs -chown -R hdfs:supergroup /flink/
1
2
3
配置conf/masters文件

[root@h001 conf]# vim masters
h001:8081
h002:8081
1
2
3
配置conf/zoo.cfg文件,添加Zookeeper集群节点

[root@h001 conf]# vim zoo.cfg
server.1=h002:2888:3888
server.2=h003:2888:3888
server.3=h004:2888:3888
1
2
3
4
启动Zookeeper集群

[root@h001 flink-1.2.0]# bin/start-zookeeper-quorum.sh
1
启动Flink集群

[root@h001 flink-1.2.0]# bin/start-cluster.sh
1
经过测试kill掉其中一个jobmanager可切换主备。

2.2. YARN集群模式高可用

当运行一个高可用YARN集群时,我们不需要运行多个JobManager(ApplicationMaster)实例,只需要运行一个实例,如果失败了通过YARN来进行重启
        Flink部署在Yarn上,仅作为yarn上“多租户”的一个service而存在。Flink在yarn中容器的概念分为2种:

用于启动JobManager(AM)的容器
用于启动TaskManager的容器
1
2
通过yarn-session.sh –help来看下启动Flink On Yarn的参数信息

其中-n代表taskmanager的容器数量,而不是taskmanager+jobmanager的容器数量
在配置HA前,先通过-q看一下我的yarn集群的资源情况:

从图中可以看出,我配置的每个NodeManager的内存是2048MB(yarn-site.xml),每个NodeManager的vcores数量是2。所以,当前yarn集群中可用内存总量为6144,总cores是6

2.2.1. FLINK ON YARN HA 配置

配置准备

在配置Flink On Yarn之前,必须保证hdfs和yarn都已经开启,可以通过HADOOPHOME/sbin/start−all.sh启动hdfs和yarn配置(yarn−site.xml)此配置需要在HADOOPHOME/sbin/start−all.sh启动hdfs和yarn配置(yarn−site.xml)此配置需要在HADOOP_CONF_DIR 的yarn-site.xml添加

[root@h001 ~]# cd /usr/bigdata/hadoop/etc/hadoop/
[root@h001 hadoop]# vim yarn-site.xml
<property>
<name>yarn.resourcemanager.am.max-attempts</name>
<value>4</value>
</property>
1
2
3
4
5
6
此配置代表application master在重启时,尝试的最大次数

[root@h001 hadoop]# clush -v -w h[002-005] --copy yarn-site.xml --dest /usr/bigdata/hadoop/etc/hadoop/
1
配置(flink-conf.yaml),此参数需要在$FLINK_HOME/conf 的flink-conf.yaml中配置

[root@h001 conf]# vim flink-conf.yaml
yarn.application-attempts: 10
1
2
      此参数代表Flink Job(yarn中称为application)在Jobmanager(或者叫Application Master)恢复时,允许重启的最大次数。
      注意,Flink On Yarn环境中,当Jobmanager(ApplicationMaster)失败时,yarn会尝试重启JobManager(AM),重启后,会重新启动Flink的Job(application)。因此,yarn.application-attempts的设置不应该超过yarn.resourcemanager.am.max-attemps

[root@h001 conf]# clush -v -w h[002-005] --copy flink-conf.yaml --dest /usr/bigdata/flink-1.5.1/conf/
1
配置zookeeper信息
      虽然flink-on-yarn cluster HA依赖于Yarn自己的集群机制,但是Flink Job在恢复时,需要依赖检查点产生的快照,而这些快照虽然配置在hdfs,但是其元数据信息保存在zookeeper中,所以我们还要配置zookeeper的HA信息。其中,recovery.zookeeper.path.namespace也可以在启动Flink on Yarn时通过-z参数覆盖。
      在yarn模式下,jobmanager.rpc.address不需要指定,因为哪一个容器作为jobManager由Yarn决定,而不由Flink配置决定;taskmanager.tmp.dirs也不需要指定,这个参数将被yarn的tmp参数指定,默认就是/tmp目录下,保存一些用于上传到ResourceManager的jar或lib文件。parrallelism.default也不需要指定,因为在启动yarn时,通过-s指定每个taskmanager的slots数量。

完整的Flink配置信息如下:

root@h001 conf]# vim flink-conf.yaml
env.java.home: /usr/java/jdk1.8.0_111
recovery.mode: zookeeper
recovery.zookeeper.quorum: h002:2181,h003:2181,h004:2181
recovery.zookeeper.path.root: /flink
recovery.zookeeper.path.namespace: /cluster_yarn
state.backend: filesystem
state.backend.fs.checkpointdir: hdfs://h001:9000/flink/checkpoints
recovery.zookeeper.storageDir: hdfs://h001:9000/flink/recovery
taskmanager.network.numberOfBuffers: 64000
fs.hdfs.hadoopconf: /usr/bigdata/hadoop/etc/Hadoop
1
2
3
4
5
6
7
8
9
10
11
以上的yarn HA配置可在Standalone集群模式下进一步添加几个参数即可完成。

2.2.2.启动FLINK YARN SESSION

在YARN上启动一个Flink主要有两种方式:

(1)、启动一个YARN session(Start a long-running Flink cluster on YARN)
(2)、直接在YARN上提交运行Flink作业(Run a Flink job on YARN)
1
2
Flink YARN Session

启动Flink Yarn Session有2种模式:

(1)、分离模式
(2)、客户端模式
1
2
      通过-d指定分离模式,即客户端在启动Flink Yarn Session后,就不再属于Yarn Cluster的一部分。如果想要停止Flink Yarn Application,需要通过yarn application -kill 命令来停止
      这种模式下会启动yarn session,并且会启动Flink的两个必要服务:JobManager和TaskManagers,然后你可以向集群提交作业。同一个Session中可以提交多个Flink作业。需要注意的是,这种模式下Hadoop的版本至少是2.2,而且必须安装了HDFS(因为启动YARN session的时候会向HDFS上提交相关的jar文件和配置文件)。我们可以通过./bin/yarn-session.sh脚本启动YARN Session。
      在启动的是可以指定TaskManager的个数以及内存(默认是1G),也可以指定JobManager的内存,但是JobManager的个数只能是一个。
采用客户端模式来启动Flink Yarn Session:

[root@h001 flink-1.5.1]# bin/yarn-session.sh -n 4 -jm 4096 -tm 8192 -s 2 -nm FlinkOnYarnSession -d -st
或者
./bin/yarn-session.sh -n 4 -tm 8192 -s 8
1
2
3
参数说明:

-n:--container 指YARN container分配的个数(即TaskManagers的个数)
-jm:--jobManagerMemory 指JobManager Containe的内存大小,单位为MB
-tm:--taskManagerMemory 指每个TaskManagerContainer的内存大小,单位为MB
-s :指每个TaskManager的slot个数
1
2
3
4
可以通过yarn的webUI查看一下当前启动的Application

通过ApplicationMaster tracking一下Flink的WebUI
http://192.168.xxx.xxx:8088/proxy/application_1500340359200_0002/#/overview
提交作业
使用bin/flink脚本提交作业,同样我们来看看这个脚本支持哪些参数:

[root@h001 flink-1.5.1]# bin/flink run
bin/flink run ./examples/batch/WordCount.jar \
--input hdfs:///user/test/LICENSE \
--output hdfs:///user/test/result.txt
1
2
3
4
后面相应的跟上参数提交作业即可。

Run a single Flink job on YARN

上面的YARN session是在Hadoop YARN环境下启动一个Flink cluster集群,里面的资源是可以共享给其他的Flink作业。我们还可以在YARN上启动一个Flink作业。这里我们还是使用./bin/flink,但是不需要事先启动YARN session:

[root@h001 flink-1.5.1]# bin/flink run -m yarn-cluster -yn 2 ./examples/batch/WordCount.jar \
--input hdfs:///user/test/LICENSE \
--output hdfs:///user/test/result.txt
1
2
3
      上面的命令同样会启动一个类似于YARN session启动的页面。其中的-yn是指TaskManager的个数,必须指定。
---------------------
作者:独行夏
来源:CSDN
原文:https://blog.csdn.net/u013368491/article/details/81610672
版权声明:本文为博主原创文章,转载请附上博文链接!

上一篇:github 搜索


下一篇:HOWTO: Be more productive