Flink on Yarn - CDH5

2020-07-10 12:01:26 浏览数 (1)

本篇文章大概2892字，阅读时间约8分钟

根据Flink官网资料，实操CDH5.16.2上配置Flink on yarn，给出了flink on yarn的集成方式和HA的配置方式

1

文档编写目的

  • 记录Flink on yarn的部署流程
  • 集群环境：CDH5.16.2

2

Flink如何与YARN交互

上图来自Flink官方

Flink YARN Client需要获取Hadoop的配置访问到集群的YARN Resource Manager和HDFS，可以使用如下方式进行配置：

  • 检查一下参数是否配置，只要有一个配置就可以读到集群的配置文件
    • YARN_CONF_DIR
    • HADOOP_CONF_DIR
    • HADOOP_CONF_PATH
  • Flink YARN Client使用HADOOP_HOME环境变量，如果设置了HADOOP_HOME，Flink YARN Client会去读取目标路径下的配置$HADOOP_HOME/etc/hadoop(Hadoop 2.X)

整个交互流程

  • 启动一个新的Flink YARN会话时，Client会先检查所请求的资源（AM的cpu和内存）是否可用。之后，会上传包含Flink和相关配置的jar包到hdfs上
  • 向YARN的RM申请容器，启动AM
  • JobManager和AM会运行在同一容器中。成功启动后，AM就会知道JobManager的地址，然后为TaskManager生成一个新的配置文件以连接到JobMananger。新生成的文件也会上传到HDFS上。
  • AM容器还会提供Flink的web界面
  • YARN分配的端口都是临时端口，因此可以同时并行执行多个Flink YARN Session
  • 最后，AM开始为TaskManager分配容器，完成以上步骤，就可以开始运行Flink的Jobs

3

在CDH5上部署Flink

添加HADOOP_CONF_DIR环境变量

代码语言：javascript复制
vi /etc/profile
# 配置HADOOP_CONF_DIR
export HADOOP_CONF_DIR=/etc/hadoop/conf
# 刷新环境变量
source /etc/profile

配置HADOOP的CLASSPATH

代码语言：javascript复制
vi /etc/profile
# 配置hadoop的classpath
export HADOOP_CLASSPATH=`hadoop classpath`
# 刷新环境变量
source /etc/profile

添加操作HDFS的用户

编辑/bin目录下的yarn-session.sh文件，配置HADOOP_USER_NAME=hdfs

代码语言：javascript复制
# 操作hdfs的用户
export HADOOP_USER_NAME=hdfs
启动Flink on yarn集群
基础功能测试

per-job提交模式

直接向yarn上提交一个example包下的wordcount任务, per job方式

代码语言：javascript复制
./bin/flink run -m yarn-cluster ./examples/batch/WordCount.jar

yarn-session提交模式

启动yarn-session

代码语言：javascript复制
./bin/yarn-session.sh -tm 8192 -s 4

yarn上生成的yarn-session对应的机器是dn2.develop.com:8042

yarn id：application_1589377449274_0645

提交任务到该yarn-session上

代码语言：javascript复制
./bin/flink run -m yarn-cluster -yid application_1589377449274_0645 ./examples/batch/WordCount.jar

进入yarn-session的web，发现任务执行完成

Flink YARN Session

利用bin下的yarn-session.sh启动yarn-session

代码语言：javascript复制
./bin/yarn-session.sh

命令行参数：

代码语言：javascript复制
Usage:
   Optional
     -at,--applicationType <arg>    设置yarn上应用的自定义程序类型
     -D <property=value>        动态参数
     -d,--detached           使用分离模式
     -h,--help            打出 cli help
     -id,--applicationId <arg>     把任务提交到正在运行的yarn session
     -j,--jar <arg>           flink jar的路径
     -jm,--jobManagerMemory <arg>    JobMananger内存
     -m,--jobmanager <arg>        连接到指定的JobManager
     -nl,--nodeLabel <arg>        指定yarn的节点标签
     -nm,--name <arg>           指定一个yarn应用的名称
     -q,--query             显示可用的yarn资源 cpu   内存
     -qu,--queue <arg>          指定yarn队列
     -s,--slots <arg>           每个TaskManager的slot
     -t,--ship <arg>            文件发送目录
     -tm,--taskManagerMemory <arg>   每个TaskManager容器的内存，默认值mb
     -yd,--yarndetached         不建议使用
     -z,--zookeeperNamespace <arg>   为高可用性模式创建zk的namespace

生产环境上用得比较多的是flink的per job模式。

Flink on YARN HA配置

在YARN HA情况下，Flink集群不需要多个JobManager实例，当JM出现故障的时候，yarn会尝试重启JM。具体的YARN的操作取决于YARN的版本。

配置YARN高可用

最大的Application Master重试次数

配置集群中yarn-site.xml的最大重试次数，集群的配置一般是2次

代码语言：javascript复制
<property>
  <name>yarn.resourcemanager.am.max-attempts</name>
  <value>4</value>
  <description>
    The maximum number of application master execution attempts.
  </description>
</property>

配置flink应用的重试次数 flink-conf.yaml

代码语言：javascript复制
yarn.application-attempts: 10

配置高可用的YARN-session

代码语言：javascript复制
high-availability: zookeeper
high-availability.storageDir: hdfs:///flink/ha/
high-availability.zookeeper.quorum: localhost:2181
state.checkpoints.dir: hdfs:///flink/flink-checkpoints

进入dn3机器，打印jps

直接kill掉yarnsessionclusterentrypoint, kill -9 117533

看到yarn上的attempt id增加， flink的web ui可以重新进行访问了 flink on yarn ha 测试完成

yarn node.js 面向对象编程 flink hadoop

0 人点赞