Flink on yarn (三种模式)-调节集群的核数和内存方法

FLINK on YARN

将flink的任务提交到yarn上运行

1、可以先关闭flink的独立集群

stop-cluster.sh

2、配置HADOOP_CLASSPATH

vim /etc/profile

# 增加
export HADOOP_CLASSPATH=`hadoop classpath`

source /etc/profile

3、启动hadoop

start-all.sh

提交任务到yarn上运行

1、Application Mode

Application Mode模式主要时为了让flink可以在K8S上运行

为每一个flink任务在yarn上启动一个集群，提交任务的main运行在jobmanager, 数据流程图在jobmanager中构建

每一个任务启动一个jobmanager

将项目打包上传到服务器

提交任务

flink run-application -t yarn-application -c com.wt.flink.core.Demo2Submit flink-1.0-SNAPSHOT.jar

查看任务列表

flink list -t yarn-application -Dyarn.application.id=application_1654846044068_0002

在yarn中可以点进去查看任务状态

关闭任务

flink cancel -t yarn-application -Dyarn.application.id=application_1654846044068_0002 52b325d666be767b24698f459bb5dda9

2、Per-Job Cluster Mode

为每一个flink任务在yarn上启动一个集群, 在本地构建数据流程图，再将数据流程图提交到jobmanager中运行

每一个任务启动一个jobmanager

提交任务

–detached: 客户端提交成功之后会退出

flink run -t yarn-per-job --detached -c com.wt.flink.core.Demo2Submit flink-1.0-SNAPSHOT.jar

查看任务列表

flink list -t yarn-per-job -Dyarn.application.id=application_1654846044068_0003

取消任务

flink cancel -t yarn-per-job -Dyarn.application.id=application_1654846044068_0003 86d6973d3d79a7040bfdf75b7cad88d0

3、Session Mode

先再yarn中启动一个flink的集群，再通过命令将任务提交到这个集群中运行

所有的任务共享同一个jobmanager

启动yarn session

yarn-session.sh -d

开启此命令后，可以在yarn客户端查看:

因为我们没有提交任何任务，所以点进去后界面是这样：

提交任务- 可以提交多个任务，多个任务共享同一个jobmanager

1、可以在网页中提交任务

2、可以通过命令行提交任务

flink run -t yarn-session -Dyarn.application.id=application_1654852007237_0008  -c com.shujia.core.Demo12ValueState  flink-1.0.jar

退出yarn-session

yarn application -kill application_1654846044068_0004

Application Mode: 每一个任务启动一个集群，任何和任务之前互不影响，在jobmanager中构建JobGraph

Per-Job Cluster Mode：每一个任务启动一个集群，任何和任务之前互不影响，在本地构建JobGraph 再将JobGraph 提交到jobmanager中运行

Session Mode: 通过sessIon模式提交的任务共用同一个集群（同一个jobmanager），如果有一个任务执行出了问题，可能会影响其它任务，一般Session 用来测试使用，因为占用的资源要少一点, 在提交任务时在动态申请taskmanager

在集群中读取kafka的数据

java.lang.ClassNotFoundException: org.apache.flink.connector.kafka.source.KafkaSource

需要将flink-sql-connector-kafka-1.15.0.jar 包上传到flink的lib目录下

yarn 资源不足问题

修改yarn-site.xml文件

增加配置

<property>
<name>yarn.nodemanager.resource.cpu-vcores</name>
<value>32</value>
</property>

<property>
<name>yarn.nodemanager.resource.memory-mb</name>
<value>1638400</value>
</property>

将配置文件同步到另外两个节点
scp yarn-site.xml node1:`pwd`
scp yarn-site.xml node2:`pwd`

重启yarn
stop-yarn.sh
start-yarn.sh

查看结果：

当然这只是假的，真正的核数和内存大小是根据我们给虚拟机设定的大小