企业官网建设流程全解析

最近在梳理Flink部署相关的资料，正好把flinkOnYarn这块单独拿出来写一篇完整的实操记录。这篇文章不讲虚的，直接围绕“为什么要用Yarn跑Flink”“两种提交模式怎么选”“配置参数踩的坑”“任务提交失败怎么排查”这几个大家最关心的问题展开。如果你正在准备把Flink接入已有的Hadoop集群，或者已经接入了但时不时遇到OOM、JobManager起不来这类问题，这篇文章应该能给你省下不少时间。

1. 为什么选择Flink On Yarn：统一资源调度的现实考量

先聊一个经常被忽视的问题：Flink有Standalone、Yarn、Kubernetes等多种部署方式，为什么大多数公司最终都选了Yarn？我在本地环境跑Flink已经够用了，为什么非要折腾一个ResourceManager去管理资源？

最直接的原因是：大多数公司的Hadoop集群早就存在了，HDFS、Hive、Spark跑得好好的。如果你为了跑Flink单独部署一套Standalone集群，就意味着一堆TaskManager要么闲置吃内存，要么高峰期不够用，还得专门有人去盯它的存活状态。而Yarn本身就是一套成熟的分布式资源调度框架，Flink作为Yarn的“租户”，按需申请Container、启动任务、释放资源，整条链路是自动化的。

有一个场景能非常直观地说明问题：你白天要跑批处理任务，晚上只跑实时计算任务，两类任务如果混用Standalone集群，你没法让Flink在晚上自动多占一些资源、白天自动让出来。但放在Yarn上，你只需要为实时任务设置一个队列，Yarn会根据集群整体负载动态分配资源，Flink任务结束后Container自动释放，完全不用人为干预。

1.1 Flink On Yarn解决的核心痛点

我把这类部署方式解决的核心问题总结成三点：

• 资源利用率：Yarn对集群资源做统一调度，Flink的任务和Spark的任务可以共用一套Hadoop集群，不会出现“Flink集群空转、Spark集群排队”的尴尬局面。
• 按需申请资源：Flink任务启动时向Yarn申请指定数量的TaskManager，任务结束后自动释放。相比之下，Standalone模式则需要手动维护固定数量的TaskManager，运维成本高。
• 任务隔离性：每个Per-Job模式的Flink作业都有独立的JobManager和TaskManager，互不干扰。咱们平时说的“这个任务把集群资源占满了、其他任务跑不动”，在Per-Job模式里基本不会发生，因为Yarn会通过队列和容器隔离来限制单个任务的最大资源。

1.2 部署模式横向对比

简单梳理一下三种模式的差异，方便没接触过的读者快速形成认知：

这里要特别提一句，Flink 1.15之后已经逐渐弱化了Per-Job模式，推荐使用Application模式。但很多公司的生产环境还停留在老版本，而且Per-Job的资源隔离思路仍然值得理解，所以这篇文章会以Per-Job为主讲，同时补充Application模式的演进逻辑。

2. 部署前必须搞定的环境准备与依赖检查

很多人在部署flinkOnYarn的时候一头扎进去改配置，结果JobManager一直起不来，最后发现是Hadoop环境变量没配对。这个环节我专门单独拿出来讲，因为根据实际经验，80%的部署失败都出在环境准备阶段。

2.1 硬件与软件最低要求

Flink On Yarn的本质是把Flink发行包提交到Yarn集群上运行，所以以下条件必须满足：

• 已部署Hadoop集群（Yarn和HDFS可用），Hadoop版本建议2.7及以上
• JDK版本1.8或1.11（Flink 1.11之后可以跑在Java 11上，但生产环境保守选JDK8更稳）
• Flink发行包，建议版本与Hadoop版本匹配。如果你用的是CDH这类发行版，尤其要注意Hadoop版本兼容性

检查环境时，先在任意一台能提交任务的客户端机器上执行：

hadoop version yarn node -list

如果这两条命令都能正常输出，说明客户端到Yarn集群的链路是通的。

2.2 最容易踩的坑：HADOOP_CLASSPATH没有配置

Flink On Yarn启动时，需要把Hadoop相关的classpath传给Yarn的AppMaster。如果你只设置了HADOOP_HOME，没设置HADOOP_CLASSPATH，大概率会碰到下面这种错误：

java.lang.NoClassDefFoundError: org/apache/hadoop/yarn/api/ApplicationMasterProtocol

这个问题的排查思路其实很简单，Yarn通过Flink提供的脚本去拿Hadoop依赖，拿不到就有这个报错。

最稳妥的解决方法是登录到Flink提交客户端，执行：

export HADOOP_CLASSPATH=$(hadoop classpath)

然后把这段export写进提交用户的环境变量配置文件（比如~/.bashrc或~/.bash_profile），避免每次开会话都手动设置。

2.3 确认Flink二进制包里的依赖情况

Flink官方发布的二进制包分为flink-1.x.x-bin-hadoop28-scala_2.12.tgz和flink-1.x.x-bin-scala_2.12.tgz两种。前者自带了Hadoop 2.8的客户端依赖，后者不带任何Hadoop相关jar包，需要通过HADOOP_CLASSPATH补充。

我个人的习惯是使用不带Hadoop依赖的包，通过HADOOP_CLASSPATH动态引入Hadoop依赖。这样做的原因是避免Flink自带的Hadoop jar和集群里的实际Hadoop版本产生冲突。这一点在CDH、HDP这类定制化Hadoop发行版上尤其明显，之前遇到过Flink自带的Hadoop 2.8 jar和CDH的RPC协议不兼容，导致AppMaster一直无法注册的情况。

2.4 搭建最小可用的Yarn测试环境

如果你的环境还没有Yarn，可以在单机上快速搭一个最小可用的伪分布式Hadoop环境用来测试：

<!-- core-site.xml --> <property> <name>fs.defaultFS</name> <value>hdfs://localhost:9000</value> </property> <property> <name>hadoop.tmp.dir</name> <value>/data/hadoop-tmp</value> </property>

<!-- yarn-site.xml --> <property> <name>yarn.nodemanager.aux-services</name> <value>mapreduce_shuffle</value> </property> <property> <name>yarn.nodemanager.aux-services.mapreduce_shuffle.class</name> <value>org.apache.hadoop.mapred.ShuffleHandler</value> </property>

启动命令顺序是：先启动HDFS，再启动Yarn：

$HADOOP_HOME/sbin/start-dfs.sh $HADOOP_HOME/sbin/start-yarn.sh

然后通过jps能看到NameNode、DataNode、ResourceManager、NodeManager这几个进程就说明环境OK了。

3. 两种核心运行模式的选型逻辑与使用场景

Flink On Yarn的模式选择，直接决定了你的系统在并发任务多时的表现和运维难度。我见过不少团队一开始为了省事只开了Session模式，结果线上任务越来越多，互相抢资源，最终不得不全部迁移到Per-Job。提前搞清楚模式的差异，能帮你少走这一趟弯路。

3.1 Session模式的本质：常驻集群，共享复用

Session模式启动后，Flink会向Yarn申请一个常驻的Flink集群。这个集群包含一个JobManager和若干TaskManager，之后所有提交的任务都提交到这个集群上执行。

工作方式：

Client -> Yarn Session (JobManager + TaskManager) -> 多个Job同时跑在同一个集群里

Session模式的优点很明显：

• 提交任务速度快，因为JobManager已经启动了，不需要每次重新申请资源
• 适合任务数量多但单个任务很小的场景，比如即席查询、小批量的SQL分析
• 运维上只需要维护一个集群，管理简单

缺点也致命：

• 所有任务共享同一个TaskManager，一个任务的OOM可能影响同集群的其他任务
• 任务之间的资源隔离性差，大任务可以把整个Session集群资源耗尽
• TaskManager挂了，上面所有任务都受影响，恢复成本高

启动Session模式的命令：

yarn-session.sh -n 2 -tm 2048 -s 2 -d

参数含义：

• -n：申请TaskManager数量
• -tm：每个TaskManager内存（MB)
• -s：每个TaskManager的Slot数
• -d：后台运行

3.2 Per-Job模式的本质：一任务一集群，隔离彻底

Per-Job模式每次提交任务时，都会向Yarn申请一个全新的Flink集群，任务结束后整个集群自动释放。这也是很多生产团队最终选择的模式。

工作方式：

Client 提交任务 -> 向Yarn申请AppMaster -> 启动JobManager -> 申请TaskManager -> 运行任务 -> 结束后释放集群

Per-Job模式的优势：

• 不同任务拥有完全独立的JobManager和TaskManager，互不影响
• 任务级别的资源隔离性好，即使某个任务OOM也不会影响其他任务的运行
• 任务结束自动释放资源，对集群资源利用率更友好

代价是任务启动时多了一步“申请资源和启动AppMaster”的过程，所以任务拉起时间比Session模式稍长，一般会多个10秒到30秒不等。

提交Per-Job任务的命令示例：

flink run -m yarn-cluster \ -yn 2 \ -ytm 2048 \ -ys 2 \ -yjm 1024 \ -c com.demo.WordCountJob \ /opt/flink-job/my-flink-demo.jar

需要注意，-m yarn-cluster表示以Per-Job模式提交，这个参数在Flink 1.11之后依然兼容老版本，但新环境建议直接使用-t yarn-per-job或干脆走Application模式。

3.3 Application模式：补上了Per-Job的最后一块短板

很多人在生产环境中会遇到这样一个问题：Per-Job模式下，用户代码是在Client端执行的，如果用户的main方法里面做了大量本地计算或者连接了外部数据库，Client端的内存和网络都可能成为瓶颈。Application模式就是为了解决这个问题而出现的。

Application模式同样是一任务一集群，但用户代码的执行位置从Client移动到了集群端的JobManager。这意味着：

• 客户端提交完任务后可以立即退出，不占用本地资源
• 任务的启动开销比Per-Job更小
• 适合长时间运行的流任务，不会因为Client断网或重启导致任务提交失败

提交Application模式的命令：

flink run-application -t yarn-application \ -Djobmanager.memory.process.size=1024m \ -Dtaskmanager.memory.process.size=2048m \ -c com.demo.StreamingJob \ /opt/flink-job/my-streaming-job.jar

3.4 如何根据实际场景选型

我根据自己的实战经验，给出一个相对实用的选型参考：

我在实际生产中最常用的是Per-Job和Application两种模式。如果你们团队还处在摸索阶段，我建议先跑Session把业务逻辑验证通了，再切到Per-Job或Application进行生产部署。

4. 实践环节：从配置到提交的完整操作步骤

这一节直接从零开始，把Flink On Yarn的完整启动流程、配置参数含义和提交步骤跑一遍，大家照着做就能搭起来。

4.1 准备Flink配置文件

找到$FLINK_HOME/conf/flink-conf.yaml，核心配置如下：

# JobManager内存配置 jobmanager.memory.process.size: 1024m # TaskManager内存配置 taskmanager.memory.process.size: 2048m # 每个TaskManager的Slot数 taskmanager.numberOfTaskSlots: 2 # 并行度默认值 parallelism.default: 2 # 是否在任务结束后保留Yarn集群（Per-Job模式可关） yarn.per-job.keep.containers: false # 自定义Yarn队列 yarn.application.queue: default

这几项从字面上不难理解。重点说一下taskmanager.memory.process.size，这是TaskManager整个进程的内存上限，包含了JVM堆内存和堆外内存。如果给TaskManager分配了4GB的容器（Container），这个值建议设置为3GB到3.5GB左右，留出一些空间给Native层和系统开销，否则容器容易超限被杀。

4.2 提交Yarn资源队列与容器的关系

Yarn中的资源调度以Container为单位，Flink的TaskManager在启动时会向Yarn申请指定大小的Container。如果你在flink-conf.yaml中设置了taskmanager.memory.process.size=2048m，那Yarn分配的Container大小也应该是2GB，二者是对应关系。

还有一个容易被忽略的参数是yarn.nodemanager.pmem-check-enabled，这个参数决定NodeManager是否检查容器物理内存使用量。如果Yarn集群设置了严格的物理内存检查，而Flink写入了大量堆外内存（比如RocksDB状态后端），就可能触发Container被杀。实际解决思路是：

• 合理设置TaskManager内存，预留20%左右作为堆外预留
• 或者调整taskmanager.memory.off-heap.size手动预留堆外空间
• 如果确实需要大堆外内存，可以在Yarn层面适当放宽物理内存检查的阈值

4.3 启动Session集群并提交测试任务

先启动Session集群：

cd $FLINK_HOME/bin # 申请2个TaskManager，每个2048MB内存，每个2个Slot yarn-session.sh -n 2 -tm 2048 -s 2 -d

执行后日志会输出类似这样的信息：

2024-01-15 14:32:45,145 INFO org.apache.flink.yarn.YarnClusterDescriptor - YARN session cluster started 2024-01-15 14:32:45,150 INFO org.apache.flink.yarn.YarnClusterDescriptor - JobManager Web Interface: http://node01:8088/proxy/application_1612345678910_0001/

然后往Session集群提交任务：

flink run -m yarn-cluster \ -p 4 \ -c com.demo.WordCountJob \ /opt/flink-job/my-flink-demo.jar \ --input hdfs:///data/input.txt \ --output hdfs:///data/output

这里使用了-p 4指定并行度。在Session模式下，如果并行度超过了TaskManager总的Slot数（上面配置是2x2=4个Slot），任务会一直等待资源而不是直接报错，这一点需要注意。并行度设置过大的情况下，提交任务看似成功，但Task一直处于SCHEDULED状态，直到有TaskManager释放出Slot。

4.4 提交Per-Job模式任务

Per-Job模式不需要预先启动Session集群，直接通过flink run -m yarn-cluster提交即可：

flink run -m yarn-cluster \ -yn 2 \ -ytm 2048 \ -yjm 1024 \ -ys 2 \ -p 4 \ -c com.demo.WordCountJob \ /opt/flink-job/my-flink-demo.jar \ --input hdfs:///data/input.txt \ --output hdfs:///data/output

这几个参数逐一解释下：

• -yn：申请TaskManager的个数，这里申请2个
• -ytm：每个TaskManager的内存，单位MB
• -yjm：JobManager的内存，单位MB
• -ys：每个TaskManager的Slot数
• -p：任务并行度

任务提交后，可以通过Yarn的Web UI观察Application的启动状态。你会先看到状态为ACCEPTED，然后JobManager启动完成后变为RUNNING。任务结束后，Per-Job模式的Application会变为FINISHED或KILLED，容器自动释放。

4.5 日志查看小技巧

在Yarn上跑Flink，最烦的就是不知道日志在哪。推荐的做法是先通过Yarn的Web UI找到Application ID，然后命令行查看日志：

yarn logs -applicationId application_1612345678910_0001

这个命令能拿到完整的容器日志，包括JobManager和TaskManager的启动日志、用户日志。如果是排查ClassNotFound或OOM这类问题，这是最快定位的方式。另外，也可以通过Flink的Web UI直接看TaskManager的日志，但前提是AppMaster还活着。

5. 生产环境中的常见坑与完整排查链路

这部分可能是全文最有实战价值的一块，因为这些坑基本都是在生产环境折腾了几个小时才总结出来的。

5.1 问题一：TaskManager启动后反复退出，容器被回收

现象描述：任务提交成功，JobManager起来了，但TaskManager一直起不来。Yarn的界面上能看到TaskManager的Container状态反复在RUNNING和KILLED之间切换。

排查链路：

第一步，查看NodeManager日志或通过yarn logs获取TaskManager容器日志：

yarn logs -applicationId application_xxx | grep -i error

第二步，重点查看日志中是否有Container is running beyond virtual memory limits或Container killed by the node manager等字样。

第三步，如果确认是物理内存超限问题，说明TaskManager进程实际内存超出Container限制，调整taskmanager.memory.process.size，或者在flink-conf.yaml中手动增加堆外内存缓冲。

我遇到过最典型的情况是：给TaskManager配了2048m内存，但状态后端使用的是RocksDB，RocksDB本身会使用大量堆外内存，最终实际占用超过2GB，Container被NodeManager杀掉。解决方案是给taskmanager.memory.off-heap.size预留至少512MB到1GB的空间。

5.2 问题二：Per-Job模式提交任务时一直卡在“Restarting”

现象描述：任务提交上去，日志显示JobManager在反复重启，查看Yarn界面发现AppMaster一直处于ACCEPTED或SUBMITTED状态。

排查链路：

这种问题大多是Yarn队列资源不足导致的。Per-Job模式每次提交任务，都会尝试在指定队列中申请一个AppMaster容器。如果队列的资源上限已经被其他任务占用，新的任务就会一直在队列中等待。

验证方式：登录Yarn ResourceManager的页面，查看目标队列的“Used Resources”和“Max Resources”。如果已用资源达到了上限，问题基本就定位了。

解决方案有几种：

• 将任务提交到资源空闲的队列
• 降低-yjm和-ytm的配置值，让任务能在剩余资源中跑起来
• 如果公司有多个计算团队，建议为每个团队配置独立的Yarn队列，避免互相占用资源

5.3 问题三：类冲突导致作业初始化失败

现象描述：任务提交成功，但在JobManager的日志中看到NoClassDefFoundError或ClassNotFoundException，大多涉及到Hadoop或第三方依赖。

排查链路：

第一种情况是缺少Hadoop依赖。解决办法是配置HADOOP_CLASSPATH，并把相关依赖加入环境变量。

第二种情况是依赖冲突，常见于Flink自带的Hadoop jar与集群的Hadoop版本不一致。处理思路是尽量使用不带Hadoop依赖的Flink二进制包，或者把Flink里与Hadoop冲突的jar包替换掉。

第三种情况是用户代码里引入了与Flink内部类冲突的库，比如引入了老版本的commons-collections或guava。排查方法是在本地用mvn dependency:tree分析依赖树，找到冲突的jar，然后在pom里做exclusion排除。

5.4 问题四：中文乱码与时区问题

在Yarn容器中跑Flink，还会遇到一个不起眼但很烦的问题：任务日志或输出结果出现中文乱码，或者时间差8小时。

这通常是因为Yarn容器运行时没有继承客户端的LANG和TZ环境变量。解决办法是在启动脚本或Flink配置中显式设置：

export LANG=zh_CN.UTF-8 export TZ=Asia/Shanghai

如果提交任务时用的是Flink SQL，建议在代码里显式设置时区：

SET table.local-time-zone = Asia/Shanghai;

5.5 历史服务器：不要忽略的运维必备组件

Yarn模式还有一个运维层面的细节——Flink HistoryServer。Per-Job模式下，任务结束后整个Yarn集群会被回收，任务运行时的指标、日志、Event信息随之消失。如果不配置HistoryServer，你事后就看不到这个任务当时的运行情况了。

配置方法：

# conf/flink-conf.yaml historyserver.web.address: 0.0.0.0 historyserver.web.port: 8082 historyserver.archive.fs.dir: hdfs:///completed-jobs/

然后启动：

$FLINK_HOME/bin/historyserver.sh start

配置完成后，任务结束后，JobManager会把Event文件写入HDFS的/completed-jobs/目录，HistoryServer会定期加载这些文件供你查询。这个功能尤其是排查“昨天跑失败的任务”时必不可少。

6. 内存参数调优的实用模板

既然讲到了flinkOnYarn，内存参数的配置是很多新手最容易看懵的地方。我把自己常用的一套生产模板直接贴出来，并说明每个参数的定位，方便大家作为基线参考。

6.1 各内存模块的角色定位

Flink进程内存大致分为三块：

• JVM堆内存（Heap）：主要用于用户代码、状态存储、网络缓冲区中的部分数据
• 托管内存（Managed Memory）：主要用于RocksDB状态后端、排序缓冲区等Flink内部数据
• 堆外内存（Direct Memory + Native）：用于网络缓冲、框架内部开销

这三个模块的关系可以简单类比为：堆内存是你租的房子里的客厅，托管内存是卧室，堆外内存是阳台。客厅小了住得挤，卧室小了放不下床，阳台完全没有衣服就没地方晾。三块都需要有合适的大小。

6.2 典型配置参考

对于单个TaskManager：

如果是使用RocksDB状态后端的流任务，建议把taskmanager.memory.managed.fraction调高到0.3左右，给RocksDB留出足够的空间。

对于JobManager：

JobManager主要负责调度和Checkpoint协调，一般不需要太大的内存。

6.3 关于容器大小与内存的关系

在Yarn上，容器分配的内存大小必须大于等于Flink进程的内存需求。如果taskmanager.memory.process.size=3072m，那么Yarn申请的Container大小最好设置为3GB或3.5GB。

我见过有人配的时候只设置了taskmanager.memory.process.size但忽略了yarn.containers.vcores，导致CPU资源分配不足，任务运行速度很慢。建议在提交时配合-yn和-ys时，也要注意每个容器的vCores配置，默认是1，如果任务计算量大，可以适当调高。

7. 从Per-Job迁移到Application模式：一次真实的上线经历

最后用一段个人经历来补充说明为什么我后来在生产环境更推荐Application模式。

早前处理一个实时ETL任务，上游数据量在高峰期会翻好几倍，任务需要较长的启动时间。一开始用的是Per-Job模式，某天因为客户端所在的机器由于网络策略调整，导致Client到Yarn集群的连接中断，Per-Job模式的任务提交过程因为Client端main方法还在执行，结果整个作业提交失败。如果是流任务重启，这一断就丢了半小时的数据。

后来把该任务迁移到Application模式后，Client端只负责上传jar包和触发提交，随后的流程全部由集群端完成，本地断网不再影响任务运行。对于需要长时间稳定运行的流任务，这种模式在运维上能省很多精力。

迁移的大致步骤：

1. 确认Flink版本在1.11以上
2. 将提交命令从flink run -m yarn-cluster改为flink run-application -t yarn-application
3. 将-yn等参数改为通过-D传参，例如-Dtaskmanager.numberOfTaskSlots=2
4. 测试本地的提交脚本，确认参数是否生效

flink run-application -t yarn-application \ -Djobmanager.memory.process.size=1024m \ -Dtaskmanager.memory.process.size=3072m \ -Dtaskmanager.numberOfTaskSlots=2 \ -Dparallelism.default=6 \ -Dyarn.application.queue=realtime \ -c com.demo.RealtimeETLJob \ /opt/flink-job/realtime-etl.jar

Application模式下的日志查看方式和Per-Job模式一致，仍然使用yarn logs -applicationId来获取，这一点没有变化。

最后再多说一句Flink On Yarn的版本选择建议。如果你现在才开始搭建，尽量选择1.14以上的版本，因为Per-Job和Session模式的代码路径逐渐收敛，Application模式愈发成熟。如果公司已有老版本作业，建议按作业维度逐步迁移，不要一次全部切到Application，毕竟每种模式的资源申请流程和故障恢复细节存在差异。我自己的习惯是按“批处理用Per-Job、长时间流任务用Application”这个思路分流，运维起来会比较清爽。