1. Flink On Yarn的部署思路与核心机制
1.1 为什么选择YARN作为Flink的部署底座
如果你在公司里跑过Flink任务，大概率不会裸奔式地在自己电脑上启动一个Standalone集群来干活。生产环境里最常见的形态，就是把Flink作业提交到已经存在的Hadoop YARN集群上，让YARN统一管理资源。这套方案叫Flink On Yarn。
我最早接触实时计算的时候，也纠结过一个问题：Flink自己不是有Standalone集群吗？为什么还要跑到YARN上？后来在真实业务里被资源问题折腾过几次才想明白，核心原因有三点。
第一，资源利用率能拉满。一个公司通常已经有了一套7x24小时跑着的Hadoop集群，机器资源都在那儿。如果Flink再单独搞一群机器，等于在资源上重复建设。把Flink作业塞进YARN，任务多的时候动态申请容器，任务少的时候资源自动释放，别的离线任务还能抢过来用。
第二，多租户管理方便。YARN本身有队列（Queue）机制，生产、测试、不同部门之间可以划分独立的资源池。你把Flink作业交到某个队列，它就只能用这个队列分配的资源，不会出现一个任务吃光整个集群的情况。这对公司运维来说是硬需求。
第三，部署成本低。不用预先搭好一套Flink集群，只需要把Flink包解压好、配置好，然后通过命令向YARN提交即可。作业起来的时候，Flink的进程会以YARN容器（Container）的形式自动拉起。
注意：Flink On Yarn的前提是网络环境能访问Hadoop集群的ResourceManager和NodeManager端口，别忽略了集群本身的安全组和防火墙配置，这个后面踩坑部分细说。
1.2 三种部署模式怎么选：Session、Per-Job、Application
Flink On Yarn按提交方式分三种，很多人刚开始容易搞混。我用大白话拆一遍。
Session模式（会话模式）
这种模式下，你先在YARN上启动一个常驻的Flink集群，这个集群有一个JobManager和若干TaskManager。后续可以通过命令往这个常驻集群里提交多个作业。它的问题在于：一个TaskManager挂了，上面跑的所有作业都受影响；多个作业挤在一起，资源争抢也很难控制。典型的适用场景是：频繁提交小作业，且作业之间相互影响不大，比如一些临时的数据查询任务。
Per-Job模式（单作业模式）
每次提交一个作业，YARN会为这个作业单独拉起一套Flink集群，作业跑完集群自动销毁。资源隔离彻底，不会出现A作业把B作业挤死的情况。这个也是最经典的Flink On Yarn用法，提交命令里指定-m yarn-cluster即可。
Application模式（应用模式）
这个比Per-Job更进一步，把用户代码的main()方法也放到YARN容器里执行。Per-Job模式下main()是在客户端本地跑的，客户端退出可能影响任务状态；Application模式则把整个应用逻辑都挪到集群内部，客户端提交完就能干自己的事。适合那种大作业、长周期作业，也适合做平台化的统一入口。
三种模式从本质上说，区别在于：Flink集群的生命周期由谁控制、main()方法在哪里执行、作业怎么跟YARN交互。下面拿一张表格直接对比，方便你做选型。
| 对比项 | Session | Per-Job | Application |
|---|---|---|---|
| Flink集群生命周期 | 常驻，手动停止 | 随作业创建和销毁 | 随作业创建和销毁 |
| main()执行位置 | 客户端 | 客户端 | YARN容器内 |
| 资源隔离 | 共享，隔离弱 | 完全隔离 | 完全隔离 |
| 启动速度 | 快（集群已起） | 慢（需拉起整套集群） | 中等 |
| 适合场景 | 小作业、交互式查询 | 独立作业 | 大作业、平台化 |
1.3 一次Flink On Yarn提交，背后到底发生了什么
理解了模式，再弄懂一次提交的完整流程，出了问题你就知道往哪儿查。
Flink作业提交到YARN后，大致流程是这样的：
- 客户端把Flink的相关依赖、配置和用户JAR包上传到HDFS临时目录里。
- 客户端向YARN的ResourceManager申请一个容器（Container），用来启动ApplicationMaster。
- ResourceManager分配容器后，NodeManager在这个容器里启动ApplicationMaster进程。
- ApplicationMaster里实际跑的是Flink的JobManager，JobManager启动成功后，会向ApplicationMaster注册资源请求。
- ApplicationMaster继续向ResourceManager申请TaskManager所需的容器。
- 容器分配下来后，NodeManager负责启动TaskManager进程，TaskManager启动后向JobManager注册自身Slot资源。
- JobManager拿到了足够的TaskManager，就开始调度作业执行。
如果提交后迟迟见不到任务在YARN上跑起来，从第2步开始逐个排查：客户端和YARN通信是否有问题、HDFS临时目录能不能写、AM容器是否因为资源不足被卡住。
2. 打好地基：环境准备与关键配置
2.1 Flink与Hadoop版本怎么匹配才不翻车
Flink版本和Hadoop版本的兼容问题，是我见过新人踩得最多的坑。很多人下载Flink压缩包之后，兴冲冲地配好环境变量就提交，结果报错说找不到org.apache.hadoop.conf.Configuration之类的类。
Flink从某个版本以后，默认发行包不再捆绑Hadoop依赖。你需要根据集群里实际安装的Hadoop版本，额外准备对应的依赖JAR。常见有两种做法：
第一种做法：直接把Hadoop安装目录下的share/hadoop相关JAR包全部拷贝到Flink的lib目录下。这个最暴力，但要注意版本别冲突，不同版本的Hadoop客户端代码差异很大。
第二种做法：用Flink官方在opt目录下预编译好的flink-shaded-hadoop-uber-*.jar，把这个JAR复制到lib目录即可。好处是官方帮你把依赖都理顺了，坏处是这个JAR包含的Hadoop版本是固定的，如果你们的Hadoop是深度定制的，可能仍有兼容问题。
除了依赖JAR，还有一个非常重要的环境变量：HADOOP_CLASSPATH。很多人在本地直接执行yarn-session.sh能启动成功，但提交作业时老是提示找不到java.lang.ClassNotFoundException: org.apache.hadoop...，其实就是因为HADOOP_CLASSPATH没有传给远端容器。
对于Hadoop 3以上的版本，我建议直接用官方推荐的方式生成：
export HADOOP_CLASSPATH=$(hadoop classpath)然后强制在/etc/profile或Flink的conf/flink-env.sh中加入这个环境变量，保证后续无论怎么调脚本都不会丢。
2.2 flink-conf.yaml中值得重点关注的参数
部署Flink On Yarn，flink-conf.yaml里的配置不再像本地跑那样随便填。我挑几个和YARN部署强相关的参数，说明一下推荐值和原因。
内存配置类参数：
| 配置项 | 推荐值 | 说明 |
|---|---|---|
jobmanager.memory.process.size | 1600m | JobManager进程的完整内存，含堆外和JVM开销 |
taskmanager.memory.process.size | 2g或更高 | TaskManager进程的完整内存，根据数据量调 |
taskmanager.numberOfTaskSlots | 按CPU核数定，通常2-4 | 每个TaskManager能跑的并行任务数 |
资源申请类参数：
yarn.application.name：给YARN上的应用起个名字，方便运维在YARN界面上认任务。yarn.application-attempts：任务失败后最多重试几次，生产环境建议设置成合理的数字，比如3，避免因偶发网络抖动直接失败。默认是1，太小了。
提交时的临时参数：
Flink支持在提交命令中用-yjm指定JobManager内存，用-ytm指定TaskManager内存，用-ys指定每个TaskManager的Slot数。比如：
flink run -m yarn-cluster \ -yjm 1024m \ -ytm 2048m \ -ys 2 \ -c com.example.MyJob \ ./my-job.jar这些临时参数会覆盖配置文件里的值，所以实际部署时，我习惯把通用的内容写进配置文件，个别作业的特殊资源要求在命令行覆盖。
注意：YARN分配内存时还会额外加上Flink自身框架的内存开销，比如TaskManager默认会多预留JVM Overhead。如果你在YARN上看到实际申请的内存比你命令里写的多，不用慌，这是正常的。
2.3 依赖JAR包的管理技巧
Flink作业的JAR包通常包含业务代码和一些第三方库，但要注意，部分依赖是不能打进业务JAR的，否则容易产生冲突。
一种常见的做法是：把Flink本身的核心依赖用provided级别声明，不要把它们打进去。如果你用Maven构建，类似下面这样：
<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java_2.11</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients_2.11</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency>另外，Flink On Yarn提交时，用户JAR也会被上传到HDFS。如果你经常改代码，HDFS上的临时目录里会堆积很多旧JAR，虽然不会立刻出问题，但日积月累也占空间。我建议定期清理HDFS上的/.flink相关临时目录。
3. 三种模式的实操记录与参数详解
3.1 Session模式：从启动yarn-session到提交作业
Session模式的流程我拆成三步。
第一步，确保Flink的启动脚本有权限访问YARN。在Flink的conf/flink-env.sh中设置好HADOOP_HOME、HADOOP_CONF_DIR和HADOOP_CLASSPATH，然后直接启动：
./bin/yarn-session.sh -d -nm flink-session-test -jm 1024m -tm 2048m -s 2解释一下参数：-d表示分离模式，客户端启动完集群后就不再占着终端；-nm指定YARN上显示的应用名；-jm和-tm分别指定JobManager和TaskManager的内存；-s指定Slot数。
启动后，如果成功，你会看到类似下面的日志：
... 2025-05-08 10:00:00,123 INFO org.apache.flink.yarn.YarnClusterDescriptor - Found Web Interface ... of JobManager ...告诉你可以通过Web UI访问Flink的监控界面了。
第二步，往这个常驻集群里提交作业。打开一个新终端，执行：
./bin/flink run -d -c com.example.MyJob ./my-job.jar \ --input hdfs:///data/input.txt \ --output hdfs:///data/output.txt-d表示作业以分离模式提交，即使客户端退出，作业也照常运行。
第三步，作业跑完以后，如果想释放资源，可以到YARN界面上Kill掉那个Application，或者执行：
./bin/yarn-session.sh --stopSession模式最大的坑在于：多个作业共享同一个TaskManager时，某个作业的CheckPoint或内存压力会把整个集群搞挂。所以我个人很少在产线长时间使用Session模式，它更适合测试环境。
3.2 Per-Job模式的正确提交姿势
Per-Job模式在我看来才是真正符合生产习惯的模式，提交一次，跑完就消亡，非常干净。
最核心的提交命令是：
./bin/flink run \ -m yarn-cluster \ -yjm 1024m \ -ytm 2048m \ -ys 2 \ -yD yarn.application.name=my-per-job-demo \ -c com.example.MyJob \ ./my-job.jar \ --input hdfs:///data/input.txt-m yarn-cluster是关键，表示让Flink以Per-Job模式提交到YARN。如果你没有显式指定-m，Flink默认使用的是default配置，那就是往本地跑或者往flink-conf.yaml里配置的地址跑。
Per-Job模式下，Flink会在HDFS的临时目录下创建以/.flink开头的应用专属目录。这个目录里放着JobManager运行所需的JAR包、配置和用户代码。作业结束后，这些临时资源会被自动清理，但如果作业失败得比较蹊跷，目录可能残留，需要手动清。
这里要特别提醒：在Per-Job模式下，默认的yarn.application-attempts可能比较小。如果你的作业因为初始化太慢导致AM被YARN杀掉，它会直接失败，而不是重启。建议在提交命令里追加：
-yD yarn.application-attempts=33.3 Application模式：把main方法交给YARN
Application模式是近几年Flink推广度比较高的用法，原因是它解决了Per-Job模式下客户端依赖太重的问题。
使用Application模式，提交命令不一样：
./bin/flink run-application \ -t yarn-application \ -Djobmanager.memory.process.size=1024m \ -Dtaskmanager.memory.process.size=2048m \ -Dyarn.application.name=my-app-demo \ -c com.example.MyJob \ ./my-job.jar区别在于：
run-application替代了run；-t yarn-application指定了目标部署目标；- 没有
-m yarn-cluster这种语法。
Application模式下，客户端只需要把JAR包传到HDFS，然后向YARN申请一个容器启动ApplicationMaster。JobManager会在AM内部运行，然后从HDFS上拉取用户JAR并执行main()方法。客户端提交完就可以退出，不再承担任何作业调度的职责。
这意味着你在线上服务器执行完提交命令，终端可以随手关掉，任务不会受到任何影响。对于K8s化、平台化的任务管理系统来说，这种模式非常友好。
三种模式的资源配置直观对比：
提交之后到YARN界面上看，你会看到：
- Session模式：一个长时间运行的应用，里面有多个作业在并行跑；
- Per-Job模式：每次提交作业都会新冒出一个独立Application；
- Application模式：跟Per-Job在YARN界面上看起来很像，但得注意ApplicationMaster的日志里能看到用户的
main()方法执行过程。
4. 常见问题与排查技巧实录
4.1 高频问题速查表
把我在实际部署里遇到的高频问题整理成一个表格，方便你直接对照。
| 现象 | 可能原因 | 解决思路 |
|---|---|---|
| 提交时报ClassNotFoundException | HADOOP_CLASSPATH未设置 | 执行export HADOOP_CLASSPATH=$(hadoop classpath) |
| 提交时报Connection Refused | YARN端口不可达或RM地址配置错误 | 核对yarn-site.xml与网络策略 |
| 作业运行中TaskManager频繁挂掉 | 内存配置超出容器限制 | 检查taskmanager.memory.process.size与YARN容器内存上限 |
| 任务一直ACCEPTED不运行 | 队列资源不足或队列不存在 | 确认-yD yarn.queue配置的队列存在且有余量 |
| AM启动后立刻被kill | AM内存不够或超时 | 增大jobmanager.memory.process.size并检查超时参数 |
| HDFS临时目录占用暴涨 | 大量失败提交残留 | 定期清理/.flink相关目录 |
4.2 结合YARN日志定位问题
有一次我遇到过任务卡死，现象是YARN上Application显示RUNNING，但Flink作业迟迟不进入RUNNING状态。客户端一直提示等待JobManager连接。
定位思路是这样的：先去YARN的ResourceManager界面找到对应Application的日志入口，点进Logs，重点看ApplicationMaster的stdout和stderr。
如果看到类似java.lang.OutOfMemoryError: Java heap space，说明JobManager或某个容器内存不够；如果看到org.apache.flink.runtime.entrypoint.ClusterEntrypointException，说明JobManager启动失败了，紧接着会看到Cause，比如端口冲突、HDFS写权限问题等。
如果是TaskManager出现异常，可以通过Flink Web UI上的TaskManager列表找到日志，查看崩溃的堆栈。
排查问题最重要的一点：先看日志，再改配置，不要凭感觉调参。很多内存问题被反复调大，其实根因是某个第三方库在序列化时产生了超大对象。
4.3 几个被验证过的独门避坑经验
第一个经验：给Flink作业在YARN上单独划分队列。不要跟离线MapReduce任务混在一个队列里，实时任务对延迟和稳定性要求高，资源被离线任务抢占后很容易出现背压或CheckPoint超时。
第二个经验：不要在生产环境用Session模式跑重要任务。我踩过一次，某个作业频繁触发CheckPoint，结果把共享TaskManager的堆内存挤爆，连带其他作业一起失败。后来上线任务统一走Per-Job或Application。
第三个经验：Flink On Yarn的日志分散在YARN的各个NodeManager节点上。如果你在容器内排查问题不方便，可以在提交命令里追加：
-yD yarn.log-aggregation-enable=true开启日志聚合后，所有日志会收集到HDFS上，配合yarn logs -applicationId <appId>查看，能省掉到处ssh登录NodeManager的麻烦。
第四个经验：HDFS上的临时目录清理不能忽视。一次大规模测试后，我检查HDFS发现.flink目录下累积了几百GB的临时JAR和配置。这不是Flink的Bug，而是很多失败任务残留没有回收。写一个定时脚本，定期删除超过一定天数的/.flink下目录，比较有效。
4.4 资源不均衡时的调优思路
有一次我们有个作业背压特别严重，TaskManager CPU飚高，但内存还有富余。很多人第一反应是加并行度、加机器，但实际上问题可能出在KeyBy的数据倾斜。
排查思路是：打开Flink Web UI，看各个SubTask的积压记录，如果发现某个子任务积压特别多而其他子任务很闲，那就是数据倾斜。这时候增加并行度是没用的，还得从KEY的设计上打散。
调整完业务层的KEY策略后，再结合Flink On Yarn的资源参数做配套优化，比如合理设置taskmanager.numberOfTaskSlots，避免一台机器上启动过多线程导致CPU争抢。
5. 从我自己的经验里总结出的部署建议
Flink On Yarn这套方案看似简单，把包解开、配几个参数、执行一条提交命令，但真正要在生产环境跑得稳，考验的是你对资源管理和任务生命周期的把控。
我个人觉得，最值得花时间的并不是去背诵各种参数，而是先想清楚你的任务特点：是长周期的实时ETL，还是高频的临时分析？是资源敏感型，还是运行时长敏感型？先想清楚这些，再来选Session、Per-Job还是Application模式，就不会错得太离谱。
如果你所在的公司已经有了一套稳定的YARN集群，建议从Per-Job模式起步，跑熟之后再尝试Application模式和平台化封装。把Flink的包和依赖管理纳入正式发布流程，比什么都重要。
最后分享一个小技巧：提交前养成使用-yD在命令行覆盖关键配置的习惯，避免为了测试不同资源配置频繁改动配置文件。配合日志聚合功能，把每次提交前的改动记录在案，出了线上问题回滚也快。
