Flink高可用集群安装

待我称王封你为后i 2022-03-26 12:44 348阅读 0赞

# 前期准备 #

## 概述 ##

Flink利用ZooKeeper在所有运行的JobManager实例之间进行分布式协调。

ZooKeeper是Flink独立的服务，需要提前安装好。

zookeeper通过领导选举和轻量级一致的状态存储提供高度可靠的分布式协调服务。

## 配置hosts ##

> 192.168.245.104 node05  
> 192.168.245.103 node04  
> 192.168.245.102 node03  
> 192.168.245.101 node02  
> 192.168.245.100 node01

## 安装依赖 ##

jdk1.8、hadoop2.6、zookeeper3.4.5

## 环境变量 ##

> export JAVA\_HOME=/opt/jdk1.8/  
> export FLINK\_HOME=/opt/flink-1.6.0  
> export HADOOP\_USER\_NAME=hdfs  
> export HADOOP\_CONF\_DIR=/etc/hadoop/conf  
> export PATH=$PATH:$JAVA\_HOME/bin:$FLINK\_HOME/bin

## 下载flink ##

> flink-1.6.0-bin-hadoop26-scala\_2.11.tgz

# 配置Standalone Cluster #

## 安装flink ##

### 解压安装包 ###

进入安装包flink-1.6.0-bin-hadoop26-scala\_2.11.tgz所在目录，执行以下命令：

> sudo tar –zxf flink-1.6.0-bin-hadoop26-scala\_2.11.tgz -C /opt

### 配置masters ###

为了启动HA群集，需要/opt/flink-1.6.0/conf/masters文件中中配置masters信息

masters内容: 包含启动JobManager的所有主机以及Web用户界面绑定的端口。

执行一下命令：

> sudo vi /opt/flink-1.6.0/conf/masters

/opt/flink-1.6.0/conf/masters内容：

> node05:8081  
> node04:8081

### 配置slaves ###

这里配置slave节点地址，一行一个地址。

执行一下命令：

> sudo vi /opt/flink-1.6.0/conf/slaves

/opt/flink-1.6.0/conf/slaves内容：

> node03  
> node02  
> node01

### 配置flink-conf.yaml ###

为了启动flink  
HA群集，需要将以下配置属性添加到/opt/flink-1.6.0/conf/flink-conf.yaml文件中

执行一下命令：

> sudo vi /opt/flink-1.6.0/conf/flink-conf.yaml

/opt/flink-1.6.0/conf/flink-conf.yaml内容：

> high-availability: zookeeper  
> high-availability.storageDir: hdfs://node05:8020/flink/ha  
> high-availability.zookeeper.quorum: node03:2181,node02:2181,node01:2181  
> high-availability.zookeeper.path.root: /flink  
> high-availability.cluster-id: cluster\_standalone

参数说明：

high-availability (必须)

作用：设置高可用用的模式，一般使用zookeeper来做flink ha的协调服务.

ZooKeeper quorum (必须)

作用：ZooKeeper  
quorum(仲裁集)是ZooKeeper服务器的备份组，它提供分布式协调服务。

ZooKeeper root (必须)

作用：flink 在zookeeper上的根节点，下面给存放集群其他节点。

ZooKeeper cluster-id （推荐配置)

作用：cluster-id ZooKeeper节点，下面放置一个集群所需的全部协调数据。

重要说明：运行YARN群集，每作业YARN会话或其他群集管理器时，不应手动设置此值。  
在这种情况下，将根据应用程序ID自动生成一个cluster-id。  
在YARN中手动设置cluster-id将覆盖此行为。 使用-z  
CLI选项指定cluster-id，将覆盖手动配置。如果在裸机上运行多个Flink  
HA群集，则必须为每个群集手动配置单独的群集标识。

Storage directory (必须)

作用：JobManager元数据会保存在文件系统storageDir中，且只有一个指向这个状态的指针存储在ZooKeeper中。

说明：storageDir存储恢复JobManager失败所需的所有元数据。

### 启动standalone集群 ###

> sudo /opt/flink-1.6.0/start-cluster.sh

## 注册服务 ##

### 创建flink-master、flink-slave服务脚本 ###

> sudo touch /etc/rc.d/init.d/flink-master

> sudo touch /etc/rc.d/init.d/flink-slave

### 授予flink-master、flink-slave运行权限 ###

> sudo chmod +x /etc/rc.d/init.d/flink-master

> sudo chmod +x /etc/rc.d/init.d/flink-slave

### 配置flink-master、flink-slave服务脚本 ###

> sudo vi /etc/rc.d/init.d/flink-master

/etc/rc.d/init.d/flink-master内容：

> \#!/bin/bash  
> \#chkconfig:2345 20 90  
> \#description:flink-master  
> \#processname:flink-master  
> export HADOOP\_CONF\_DIR=/etc/hadoop/conf  
> export JAVA\_HOME=/opt/jdk1.8  
> FLINK\_HOME=/opt/flink-1.6.0  
> MASTER=\`cat /etc/sysconfig/network | grep HOSTNAME | awk -F= ‘\{print  
> $2\}’\`  
> WEB\_UI\_PORT=8081  
> \#start flink-master  
> start()\{  
> $FLINK\_HOME/bin/jobmanager.sh start “$\{MASTER\}” “$\{WEB\_UI\_PORT\}”  
> \}  
> \#stop flink-master  
> stop()\{  
> pid=$(ps -ef | grep StandaloneSessionClusterEntrypoint | grep -v grep |  
> awk ‘\{print $2\}’)  
> if \[ “$pid” != “” \]; then  
> kill -9 $pid  
> fi  
> echo “stopping standalonesession daemon on host $\{MASTER\}”  
> \}  
> \#status flink-master  
> status()\{  
> pid=$(ps -ef | grep StandaloneSessionClusterEntrypoint | grep -v grep |  
> awk ‘\{print $2\}’)  
> if \[ “$pid” != “” \]; then  
> echo “standalonesession daemon (pid “$pid”) is running”  
> else  
> echo “standalonesession daemon is not running”  
> fi  
> \}  
> \#restart flink-master  
> restart()\{  
> stop  
> start  
> \}  
> case $1 in  
> start) start;;  
> stop) stop;;  
> status) status;;  
> restart) restart;;  
> \*) echo “require start|stop|status|restart” ;;  
> esac

> sudo vi /etc/rc.d/init.d/flink-slave

/etc/rc.d/init.d/flink- slave内容：

> \#!/bin/bash  
> \#chkconfig:2345 20 90  
> \#description:flink-slave  
> \#processname:flink-slave  
> export HADOOP\_CONF\_DIR=/etc/hadoop/conf  
> export JAVA\_HOME=/opt/jdk1.8  
> FLINK\_HOME=/opt/flink-1.6.0  
> SLAVE=\`cat /etc/sysconfig/network | grep HOSTNAME | awk -F= ‘\{print  
> $2\}’\`  
> \#start flink-slave  
> start()\{  
> $FLINK\_HOME/bin/taskmanager.sh start  
> \}  
> \#stop flink-slave  
> stop()\{  
> pid=$(ps -ef | grep TaskManagerRunner | grep -v grep | awk ‘\{print  
> $2\}’)  
> if \[ “$pid” != “” \]; then  
> kill -9 $pid  
> fi  
> echo “stopping taskmanagerrunner daemon on host $SLAVE”  
> \}  
> \#status flink-slave  
> status()\{  
> pid=$(ps -ef | grep TaskManagerRunner | grep -v grep | awk ‘\{print  
> $2\}’)  
> if \[ “$pid” != “” \]; then  
> echo “taskmanagerrunner daemon (pid “$pid”) is running”  
> else  
> echo “taskmanagerrunner daemon is not running”  
> fi  
> \}  
> \#restart flink-slave  
> restart()\{  
> stop  
> start  
> \}  
> case $1 in  
> start) start;;  
> stop) stop;;  
> status) status;;  
> restart) restart;;  
> \*) echo “require start|stop|status|restart” ;;  
> esac

### 启动/停止/查看flink-master、flink-slave ###

> sudo service flink-master start

> sudo service flink-master stop

> sudo service flink-master status

> sudo service flink-slave start

> sudo service flink-slave stop

> sudo service flink-slave status

### 设置flink-master、flink-slave开机自启 ###

> sudo chkconfig flink-master on

> sudo chkconfig flink-slave on

# 配置YARN Cluster #

## 安装flink ##

### 解压安装包 ###

进入安装包flink-1.6.0-bin-hadoop26-scala\_2.11.tgz所在目录，执行以下命令：

> sudo tar –zxf flink-1.6.0-bin-hadoop26-scala\_2.11.tgz -C /opt

### 配置yarn-site.xml ###

此配置需要在$HADOOP\_CONF\_DIR  
的yarn-site.xml添加，HADOOP\_CONF\_DIR默认配置/etc/hadoop/conf。

> sudo vi /etc/hadoop/conf/yarn-site.xml

/etc/hadoop/conf/yarn-site.xml添加内容：

> <property>  
> <name>yarn.resourcemanager.am.max-attempts</name>  
> <value>4</value>  
> <description>  
> The maximum number of application master execution attempts.  
> </description>  
> </property>

### 配置flink-conf.yaml ###

为了启动flinkHA群集，需要将以下配置属性添加到/opt/flink-1.6.0/conf/flink-conf.yaml文件中

执行一下命令：

> sudo vi /opt/flink-1.6.0/conf/flink-conf.yaml

/opt/flink-1.6.0/conf/flink-conf.yaml内容：

参数说明：

yarn.application-attempts (必须)

作用：此参数代表Flink  
Job（yarn中称为application）在Jobmanager（或者叫Application  
Master）恢复时，允许重启的最大次数。

注意，Flink On Yarn环境中，当Jobmanager（Application  
Master）失败时，yarn会尝试重启JobManager（AM），重启后，会重新启动Flink的Job（application）。因此，yarn.application-attempts的设置不应该超过yarn.resourcemanager.am.max-attemps。

其他参数同Standalone Cluster配置中2.1.4 flink-conf.yaml参数说明。

### 启动yarn集群 ###

下面会启动yarn session，其中4 个Task Managers 进程，即yarn  
container，每个分配4 GB内存)。另外，使用-s选项指定每个Task  
Manager进程使用的processing slots数量，官方建议设置成每台机器（yarn node  
manager）的CPU processors 个数。

> sudo /opt/flink-1.6.0/yarn-session.sh -n 4 -s 4 -jm 1024 -tm 4096

参数说明：

\-n 必须，要分配的YARN容器数（即TaskManager容器数）。

\-s 每个TaskManager的插槽数，官方建议和cpu核数相同。

\-d 启动分离模式，即客户端在启动Flink Yarn Session后，就不再属于Yarn  
Cluster的一部分。如果想要停止Flink Yarn Application，需要通过yarn  
application -kill <Application\_ID>命令来停止。

\-q 显示可用的YARN资源（内存，内核）。

\-nm 在YARN上为应用程序设置自定义名称。

\-qu 指定YARN队列。

\-st 以流模式启动Flink

\-jm JobManager容器的内存\[以MB为单位\]。

\-tm TaskManager每个容器的内存\[以MB为单位\]。