第三章 Flink基础理论之运行模式

• 运行模式：独立集群（standlone模式）和on yarn模式
• STANDLONE模式：Flink程序运行在Flink独立集群上，计算资源不依赖于外部。
• Flink On Yarn模式：Flink程序通过Flink客户端提交到Yarn队列上，计算资源依赖于Yarn上

1、Standlone模式

1.1、集群搭建配置

• conf/master配置JobManager地址：【hadoop102:8081】
• conf/workers配置TaskManager机器地址：【hadoop103/hadoop104】
• conf/flink-conf.yaml：【程序参数配置】

1.2、集群启停命令

#独立集群启动命令
bin/start-cluster.sh
#独立集群停止模式
bin/stop-cluster.sh

1.3、应用提交

①提交方式：通过Web Ui提交Job

②提交方式：通过flink run提交Job

./flink run ../examples/batch/WordCount.jar --input /opt/flink/examples/data/123.txt --output /opt/flink/examples/data/test.txt

1.4、standlone模式缺点

①资源利用弹性不够：TaskManager数目固定，每个TaskManagerSlot数量固定，故总的Slot数量固定。
②资源隔离度不够：所有Job共享集群的资源，一个Job出现故障可能会影响另一个Job。
③所有Job共用一个jobmanager负载过大。

2、Flink On Yarn模式

• 本质：本质上去Yarn集群上申请容器，来运行flink的jobmanager+taskmanager集群。

根据集群的生命周期、资源的隔离保证、Main运行方式分为以下三种模式：

2.1、Session模式

• 本质：多个Job共享一个集群、Job退出集群也不会退出【除非在Yarn上Kill掉】、Main方法运行在客户端。

• 适用场景：需要频繁提交大量小Job场景比较适用【每次提交job时不需要向Yarn注册应用】

• 缺点：预先在yarn上启动一个flink集群，然后将任务提交到集群上，这种模式集群中任务使用相同的资源，如果某一个任务出现了问题导致整个集群挂掉，那就得重启集群中的所有任务

①第一步：启动集群

bin/yarn-session.sh -jm 2048 -tm 2048 -s 1 -m yarn-cluster -nm flink_test -qu default

# jm: jobmanager memory
# tm: taskmanager memory
# -m yarn-cluster: 集群模式(Yarn集群模式)
# -s: 规定每个taskmanager上的taskSlot数
# -nm: 自定义application名称
# -qu: 指定要提交到的yarn队列

• 注意事项：Flink集群本质上是一个分布式程序，当运行在Yarn上时就是一个application。
  • 如果是一个MR程序：对于Yarn来说一个Application，具体进程包括【MrApplicationMaster、yarnchild】
  • 如果是一个Spark程序：对于Yarn来说是一个Application，具体进程包括【AppMaster、CoarseGrainYarnExecutor】
  • 如果是一个Flink程序：对于Yarn来说是一个Application，具体进程包括【JobManager、TaskManager】
• 运行结果

②第二步：提交任务，即向已运行的Session模式集群提交Job

./flink run -d -yid application_1663807338362_0009 -p 2  ../examples/batch/WordCount.jar --input /tmp/123.txt  --output hdfs://leidi01:8020/test/WC1.txt

• 运行结果

2.2、Per-Job模式

• 本质：每个Job独享一个集群【即每个Job都有自己的JobMnager和TaskManager】，Job退出集群退出、Main方法运行在客户端。

• 适用场景：大Job，运行时间较长。【因为每起一个Job，都要向Yarn申请容器，耗时较长】

• 缺点：jar包的解析、生成JobGraph是在客户端上执行，然后将生成的jobgraph提交到集群

./bin/flink run \
# 指定yarn的Per-job模式，-t等价于-Dexecution.target
-t yarn-per-job \
# yarn应用的自定义name
-Dyarn.application.name=consumerDemo \
# 未指定并行度时的默认并行度值, 该值默认为1
-Dparallelism.default=3 \
# JobManager进程的内存
-Djobmanager.memory.process.size=2048mb \
# TaskManager进程的内存
-Dtaskmanager.memory.process.size=2048mb \
# 每个TaskManager的slot数目, 最佳配比是和vCores保持一致
-Dtaskmanager.numberOfTaskSlots=2 \
# 防止日志中文乱码
-Denv.java.opts="-Dfile.encoding=UTF-8" \
# 支持火焰图, Flink1.13新特性, 默认为false, 开发和测试环境可以开启, 生产环境建议关闭
-Drest.flamegraph.enabled=true \
# 入口类
-c xxxx.MainClass \
# 提交Job的jar包
xxxx.jar

• 提交示例

./flink run -t yarn-per-job ../examples/batch/WordCount.jar --input /tmp/123.txt  --output hdfs://leidi01:8020/test/WC.txt

• 运行结果

2.3、Application模式

• 核心：多个Job共享一个JobManager、Main方法运行在集群。

• 适用场景：生产环境建议使用

./flink run-application -t yarn-application ../examples/batch/WordCount.jar --input /opt/flink/examples/data/123.txt  --output /opt/flink/examples/data/wc.txt

3、常见问题

3.1、yarn-session运行报错

• 问题现象

2022-09-14 21:01:13,594 ERROR org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Error while running the Flink session.
java.lang.NoClassDefFoundError: javax/ws/rs/ext/MessageBodyReader
        at java.lang.ClassLoader.defineClass1(Native Method) ~[?:1.8.0_121]
        at java.lang.ClassLoader.defineClass(ClassLoader.java:763) ~[?:1.8.0_121]

• 问题原因：缺少依赖
• 解决方案：下载相关jar包

https://link.csdn.net/?target=https%3A%2F%2Frepo1.maven.org%2Fmaven2%2Fjavax%2Fws%2Frs%2Fjavax.ws.rs-api%2F2.0%2Fjavax.ws.rs-api-2.0.jar

3.2、yarn-session运行报错

• 问题现象：

 Error while running the Flink session.
org.apache.flink.configuration.IllegalConfigurationException: JobManager memory configuration failed: The configured Total Process Memory size (1024.000mb (1073741824 bytes)) is less than the sum of the derived Total Flink Memory size (1024.000mb (1073741824 bytes)) and the configured or default JVM Metaspace size  (256.000mb (268435456 bytes)).

• 解决方案：扩大内存

./yarn-session.sh -jm 2048 -tm 1688 -s 2 -m yarn-cluster -nm flink_test -qu default

• 二次问题

Caused by: org.apache.flink.configuration.IllegalConfigurationException: The number of requested virtual cores per node 2 exceeds the maximum number of virtua l cores 1 available in the Yarn Cluster. Please note that the number of virtual cores is set to the number of task slots by default unless configured in the Flink config with 'yarn.containers.vcores.'

• 解决方案：调整内存槽

./yarn-session.sh -jm 2048 -tm 2048 -s 1 -m yarn-cluster -nm flinksql -qu default

• 三次问题

Caused by: org.apache.hadoop.security.AccessControlException: Permission denied: user=root, access=WRITE, inode="/user":hdfs:hdfs:drwxr-xr-x

• 解决方案：切换HDFS用户提交命令 【设置dfs.permissions.enabled=false】

su hdfs
./yarn-session.sh -jm 2048 -tm 2048 -s 1 -m yarn-cluster -nm flink_test -qu default

3.3、Flink on Yarn集群启动报错

• 报错现象

Deployment took more than 60 seconds. Please check if the requested resources are available in the YARN cluster

• 问题原因：yarn资源不足
• 解决方案：配置yarn-site.yml

<property>
    <name>yarn.scheduler.minimum-allocation-mbname>
    <value>1024value>
property>
<property>
    <name>yarn.scheduler.maximum-allocation-mbname>
    <value>102400value>
property>
<property>
    <name>yarn.nodemanager.resource.cpu-vcoresname>
    <value>32value>
property>
<property>
    <name>yarn.nodemanager.resource.memory-mbname>
    <value>51200value>
property>

3.4、Flink On Yarn提交任务后处于ACCEPT状态

• 问题现象

• 问题原因
  • 第一个原因可能是当前正在运行的任务数量达到了集群允许同时运行的任务数量的上限
  • 第二个原因可能是集群资源不足，任务在等待其他任务释放资源
• 解决方案：kill任务

./yarn application -kill application_1662709229333_0005