Flink on YARN集群搭建实战：版本兼容、内存模型与队列调度

我最近在给一套已有的Hadoop YARN集群接入Flink，目标是让流式计算和现有的离线任务共享同一个资源池。看着flink run和yarn-session.sh这两个入口，很多人会觉得搭个Flink on YARN集群就是解压、配配参数、跑起来。真动手之后才发现，版本兼容、内存模型、三种运行模式、队列调度这几件事如果不提前捋清楚，后面每一轮排障都会很痛苦。本文把我这次从选型到落地的完整流程梳理出来，重点写文档里不会直接告诉你的坑，适合已经有Hadoop集群、准备把Flink接进去的读者参考。

我默认你已经有一套能正常运行的Hadoop YARN集群，HDFS和YARN都在工作，接下来要做的不是“再搭一套Flink集群”，而是把Flink作为一个YARN应用嵌进去。

1. 为什么最终选择Flink on YARN，而不是给Flink单独拉起集群

1.1 资源复用是最大的动机

一开始我也纠结过要不要用Flink Standalone。Standalone模式下，需要单独留出几台机器给Flink用，JobManager和TaskManager以独立进程的方式由你自己管理。这意味着你还要额外做一套进程守护、节点监控、重启策略，机器空闲时资源也没法分给其他计算任务。

Flink on YARN则完全不同。它把JobManager和TaskManager都当成YARN上的容器来申请和运行，YARN负责调度、分配、重启。换句话说，Flink不再是一套独立集群，而是变成了YARN上的一个“常驻应用”。我现有的MapReduce、Spark任务可以继续在同一个YARN集群里跑，Flink需要多少资源就临时向YARN申请多少，用完释放，资源利用率自然就上去了。

1.2 故障恢复能力直接被继承过来

YARN对容器有完整的故障重试机制。TaskManager所在节点宕机，YARN会检测到容器失败，然后在其他存活节点上重新申请资源拉起一个新的TaskManager。JobManager如果配置了ZooKeeper等高可用方案，也能由YARN配合完成主备切换。

这一点对比Standalone优势非常明显。Standalone模式下TaskManager挂了，进程守护脚本只能做到“起来一个新手进程”，但新进程是否和JobManager正常建立连接、是否重新获取到正确的任务分片，这些都要额外处理。YARN做容器调度是它的本职工作，稳定性可靠得多。

1.3 多租户隔离比专门搭一套物理集群更现实

集群做过YARN队列规划之后，Flink任务可以放到专门的实时队列，离线任务放离线队列。不同部门、不同业务线之间互不干扰。如果你单独拉一套Flink集群，这种隔离反而很难做——你得自己实现资源配额和访问控制。

当然，Kubernetes平台在流计算场景下也很强，但那是另一条技术路线。如果你的团队还没有K8s，为Flink单独引入一套容器平台成本很高；反之，YARN已经在线上稳定运行，Flink on YARN是改动最小、见效最快的方案。

2. 版本与基础环境规划：先定Flink和Hadoop的兼容连锁清单

2.1 版本组合是怎么定下来的

版本选型是Flink on YARN搭建中最容易忽略、又最容易踩爆的一环。Flink对Hadoop客户端版本并不是“任何版本都能随便配”。我这次遇到的情况是线上为Hadoop 3.3.x，综合考虑后选了Flink 1.18，JDK用11。

有一点必须强调：新版Flink的发布包不再像老版本那样打包一份“预编译Hadoop uber jar”让你随意挑。新版Flink自身不捆绑Hadoop客户端依赖，运行时需要你显式提供HADOOP_CLASSPATH和HADOOP_CONF_DIR。如果你是从Flink 1.14及更早版本平滑迁移，习惯还在“找个对应Hadoop版本的jar丢进lib目录”这种模式，一定要改过来。

export HADOOP_HOME=/opt/hadoop export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop export HADOOP_CLASSPATH=$(hadoop classpath)

这行hadoop classpath会把Hadoop所有客户端依赖的jar路径合并成一个大字符串。不设置的话，Flink进程在运行时找不到HDFS文件系统实现类，启动报错通常是ClassNotFoundException: org.apache.hadoop.fs.FileSystem，你还不一定能第一时间想到是这个原因。

2.2 Java版本必须和YARN NodeManager保持一致

YARN容器是由NodeManager启动的，NodeManager使用哪个Java版本启动容器，容器内进程就默认使用哪个Java。如果Flink编译版本比NodeManager上的JVM版本高，任务提交后在容器启动阶段就会报UnsupportedClassVersionError。

比较稳妥的做法是：整个Hadoop集群、Flink客户端、YARN容器统一使用同一大版本的JDK。我生产用的组合是Hadoop 3.3.x + JDK 11 + Flink 1.18，实测没遇到兼容问题。如果集群还在用老JDK 8，那Flink版本就不要选太新，Flink 1.13到1.16区间配合JDK 8是常见搭配。

二一个需要注意的点：Flink打包时与Scala版本有关联。如果你不用Flink的Scala API，只写Java或者通过SQL提交作业，那选2.12还是2.13影响不大；一旦有历史Scala任务要迁移，就尽量选择和之前相同的Scala版本，避免类路径冲突。

2.3 规划一个合理的示例集群

我做资源规划时按下面这个最小规模来打样，三台机器足够跑通整套流程：

这套规划下YARN集群的实际可用资源大约从node02和node03出，node01主要承担管理角色。如果你机器多，继续增加DataNode/NodeManager节点即可，Flink不需要单独预留机器。

2.4 基础环境四件套别偷懒

正式搭建前，花十分钟检查四件事：

1. 所有节点时间同步。YARN对容器心跳有时间判断，时钟偏差过大会出现节点被ResourceManager误判为异常，进而导致容器反复调度。
2. 主机名和DNS解析。/etc/hosts要写好映射，Flink提交时HDFS地址如果用主机名，解析不了会直接失败。
3. 防火墙。至少保证ResourceManager Web端口、Flink Web UI所在端口、节点间通信端口不挡。
4. 客户端到YARN的提交权限。如果你用普通用户提交，需要提前在YARN上给这个用户或用户组授权。

3. 三种运行模式的差异拆解：Session、Per-Job、Application如何取舍

3.1 Session模式：常驻会话适合临时查询

Session模式启动一个长期驻留的YARN会话，一个JobManager放在那里，后面提交多个作业都复用它。好处是提交快、不需要每次启动新JobManager，非常适合开发调试和临时跑一些小的分析作业。

但Session模式的问题也出在“复用”上。所有任务共享同一个JobManager的资源与内存，任务多了JobManager会成为瓶颈；业务上如果有一个作业把资源占满，其他作业只能排队。而且Session本身是一个常驻进程，不及时关闭就会一直占用YARN队列资源。

3.2 Per-Job模式：一个作业一个JobManager

Per-Job模式是Flink 1.15之前生产环境最常用的模式。每次提交都会在YARN上启动一个独立的JobManager，作业结束整个集群资源随之释放，互不干扰。隔离性比Session好很多，适合长周期的正式作业。

不过Per-Job模式在提交过程中依赖客户端作为“中间人”。客户端提交作业后还要负责和JobManager通信，客户端一旦退出，作业有中断风险。这一点在Flink 1.15之后的版本里逐渐被弱化，新版本更推荐用Application模式。

3.3 Application模式：把main方法也交给YARN

Application模式从Flink 1.11开始引入，核心变化是：用户的main方法在YARN的ApplicationMaster容器里执行。你本地只要提交整个作业jar包，YARN拿到之后会在集群内启动AM，AM直接跑main方法，创建JobManager和后续的TaskManager。

这个模式解决了Per-Job依赖客户端的问题。提交命令执行完毕，本地终端就可以关了，作业照样在YARN上跑。生产环境里长周期作业我基本默认用Application模式。

3.4 我的选型建议

一句话总结：如果你在规划新集群，不用犹豫，默认Application模式。Session只留给真正有“频繁提交短任务”诉求的场景。

4. 从零开始的搭建步骤：依赖环境、配置文件到首次提交

4.1 先确认YARN集群有足够资源

动手配置Flink之前，我先在客户端验证Hadoop集群状态：

hdfs dfs -ls / yarn node -list -all yarn cluster -metrics

yarn node -list -all能看到所有节点是否Health状态，yarn cluster -metrics能看到集群总内存、可用内存、剩余核数。只有可用内存大于你即将申请的资源，作业才能顺利跑起来。

这一步别跳过，节省排查时间的效果非常明显。我有一次配置完全没问题，但作业一直ACCEPTED不启动，查到最后是集群可用内存被其他任务占了，YARN在排队等资源。

4.2 解压Flink并配置flink-conf.yaml

Flink目录结构很简单，bin目录放启动脚本，conf目录放配置文件，lib目录放依赖。拿到压缩包解压后，先改conf/flink-conf.yaml。

我给出一个能直接运行的基准配置：

jobmanager.memory.process.size: 2048m taskmanager.memory.process.size: 4096m taskmanager.numberOfTaskSlots: 4 parallelism.default: 8 jobmanager.execution.failover-strategy: region state.backend: hashmap state.checkpoints.dir: hdfs://node01:8020/flink/checkpoints yarn.application.queue: flink

解释几个关键参数：

• jobmanager.memory.process.size是JobManager整个进程的总内存，不是堆大小。Flink会在这个额度内自行划分JVM堆、元空间、网络缓冲等。
• taskmanager.memory.process.size是每个TaskManager进程的总内存，同理，不是简单对应堆。
• taskmanager.numberOfTaskSlots是每个TaskManager能同时执行的并行任务数，我给4，比较折中。
• parallelism.default是作业默认并行度。这里设8，意味着如果作业不显式指定并行度，就需要2个TaskManager实例，因为每个实例有4个slot。

4.3 配置Hadoop环境变量

在/etc/profile.d/flink.sh里写入：

export HADOOP_HOME=/opt/hadoop export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop export HADOOP_CLASSPATH=$(hadoop classpath) export JAVA_HOME=/usr/lib/jvm/java-11 export PATH=$JAVA_HOME/bin:$HADOOP_HOME/bin:$PATH

这里HADOOP_CONF_DIR不是可有可无。Flink要从这里读取core-site.xml、hdfs-site.xml、yarn-site.xml才知道HDFS地址和YARN ResourceManager地址。

Flink 1.15之后的版本如果缺失Hadoop依赖，还需要在flink-conf.yaml中配置：

env.hadoop.conf.dir: /opt/hadoop/etc/hadoop env.java.home: /usr/lib/jvm/java-11

env.java.home的作用是让Flink脚本启动的进程使用指定JDK，而不是依赖当前shell的PATH。统一在这里锁定Java版本，能避免提交时没问题、容器启动时版本不一致的尴尬。

4.4 先用Session模式快速验证通路

资源充足、配置就绪后，我先用Session模式跑一遍，验证Flink和YARN的通道是否正常：

./bin/yarn-session.sh -d -nm flink-session-test -jm 2048m -tm 4096m -s 4

-d是后台运行，-nm指定YARN应用的显示名称，-jm和-tm分别指定JobManager和TaskManager内存，-s指定每个TaskManager的slot数。

执行成功后，ResourceManager的Web UI会多出一个名为flink-session-test的应用，同时终端会打印出Flink Web UI的访问地址。

然后提交一个自带示例作业：

./bin/flink run -d -t yarn-session -p 8 ./examples/streaming/TopSpeedWindowing.jar

sbin目录下的示例jar路径不同版本略有差异，但一般能在examples/下找到Streaming任务。提交后去Flink Web UI看TaskManager数量和slot分配，如果看到2个TaskManager、总共8个slot，说明整体链路已经通了。

4.5 用Application模式跑正式作业看差异

Session模式验证完之后，再试一下生产推荐的Application模式：

./bin/flink run-application -t yarn-application -p 8 -c com.example.demo.MyFlinkJob /opt/app/flink-demo.jar

注意这里用的是run-application而不是run，后面必须跟-t yarn-application。如果漏了-t，命令可能按本地模式解释，作业不会提交到YARN上。

Application模式启动后，YARN上会出现一个和Session名字不同的应用，该应用的AM容器会直接执行MyFlinkJob的main方法。此时把本地终端直接关掉，登录YARN再yarn application -list看一眼，应用还在运行。这个特性就是它替代Per-Job模式的核心原因。

4.6 提交后的三查

作业提交后，排查前先按顺序查三样东西：

yarn application -status application_1686900000000_0012 yarn logs -applicationId application_1686900000000_0012 yarn application -list -appStates ALL

-status看的是最核心的YARN层诊断信息，-logs拿的是作业日志，第三个用来确认应用整体生命周期。学会这三个命令，Flink on YARN环境下的应用状态检查就够用了。

5. 内存模型与资源配额计算：slot、堆内堆外、容器内存的联动关系

5.1 为什么Flink要推“进程总内存”而不是只调堆

我接触过不少从老版本迁移过来的运维同学，习惯性只调taskmanager.heap.size，新版本的校验会直接跑不过，因为Flink要想清楚一件事：进程的内存需求远不只JVM堆这么大。

一次正常的流式任务运行过程中，内存会被拆成几个部分：

• JVM堆：保存算子状态、用户代码对象、内部数据缓冲。
• 托管内存（Managed Memory）：排序、缓存、以及RocksDB状态后端在堆外占用的内存。
• 网络缓冲（Network Buffer）：TaskManager各slot之间传输数据需要的内存。
• JVM Overhead：Metaspace、线程栈、JVM自身开销。

如果只把堆内存调大，堆外得不到保障，任务处理高峰一来就会OOM，容器被YARN杀掉，然后不断重启，看起来像集群不稳定，实际是内存配比不合理。

所以新版本的Flink强烈建议统一使用taskmanager.memory.process.size，把整个进程的总内存额度交给Flink自己去拆分。这样最容易保证各部分均衡，也最容易通过启动校验。

5.2 一个具体的资源计算例子

假设你有一个作业，业务压力估算之后确定并行度需要16。按前面配置，taskmanager.numberOfTaskSlots=4，那么需要的TaskManager数量就是：

TaskManager数量 = 16 / 4 = 4

如果每个TaskManager进程总内存是4G，JobManager进程内存是2G，那么需要向YARN申请的容器内存总量是：

容器内存总量 = 4 * 4G + 2G = 18G

看起来数字不大，但你得同时考虑所在YARN队列的容量。如果队列最大可用内存是16G，这个作业就要排队等待资源释放。调调并行度、降降TM内存，往往就调度上去了。

5.3 容器内存超限的坑：为什么会有物理内存超限

YARN容器有内存限制，进程实际占用的物理内存超过容器申请值，NodeManager会直接杀掉容器。Flink的进程总内存已经包含了JVM Overhead，按理说不会超，但如果某个TaskManager使用RocksDB状态后端，且managed memory分配的比例不够，RocksDB持续写磁盘前的缓存就会把堆外内存顶破。

常见的三个直接原因：

1. taskmanager.memory.jvm-overhead.fraction默认0.1，对某些状态应用来说偏小，我一般调到0.15。
2. 并发slot太多，每个slot上的子任务同时创建大量临时对象。
3. 用户代码里缓存了大的静态数据结构，比如把几十G的维度表直接加载到堆上。

排查方向很简单：先yarn logs看是不是OutOfMemoryError。如果是物理内存超限引起Kill，日志会有明确单词physical memory of container exceeds limit。

5.4 slot数到底设置多少合理

slot不是越多越好。slot越多，单个TaskManager上并发执行的子任务就越多，JVM堆内竞争、GC压力都随之上升。我一般推荐单TaskManager 2到8个slot，最常用的是4。

如果你看到某一个JobManager下挂着大量TaskManager，每个只有1个slot，虽然隔离性最好，但JVM开销占比太高，内存浪费严重。反过来一个TaskManager塞16个slot，数据处理高峰时GC几乎停不下来。

5.5 JobManager内存别给太大

JobManager在YARN容器里只做调度、检查点协调、作业图管理这些事，和数据处理没直接关系。数据压力不在它身上，所以它不需要和TaskManager同样级别的内存。

单个作业1G到2G足够，除非你的Session模式挂了几十个作业，或者Flink Web UI承载大量并发请求，才需要适当提高JobManager内存。给多了没有意义，占用的YARN坑位还平白挤压其他任务。

6. 实际运行中的资源调优与多租户隔离实践

6.1 在YARN上给Flink划一个独立队列

Flink要上生产，第一步就是别让Flink任务和离线任务混在同一个YARN队列里。离线批任务通常是时段性高占用，Flink流任务需要长期稳定资源，两者放在一起，Flink的资源很容易被批任务的资源高峰抢掉。

这里以Capacity Scheduler为例做一个最小配置，在capacity-scheduler.xml中：

<property> <name>yarn.scheduler.capacity.root.queues</name> <value>default,flink</value> </property> <property> <name>yarn.scheduler.capacity.root.flink.capacity</name> <value>40</value> </property> <property> <name>yarn.scheduler.capacity.root.default.capacity</name> <value>60</value> </property>

这样根队列下面就有了default和flink两个子队列，Flink专用队列占总资源40%。改完配置要执行yarn rmadmin -refreshQueues才能生效。

6.2 提交作业时指定队列

队列设置好之后，提交作业时用-D动态指定：

./bin/flink run-application -t yarn-application \ -D yarn.application.queue=flink \ -c com.example.demo.MyFlinkJob /opt/app/flink-demo.jar

如果你希望某个集群所有Flink作业都默认走某队列，可以在flink-conf.yaml里固定：

yarn.application.queue: flink

6.3 运行参数最好用-D在提交时覆盖

配置文件是默认值，具体作业最好在执行命令里用-D覆盖。比如某个作业数据量大，需要更多内存：

./bin/flink run-application -t yarn-application \ -D parallelism.default=32 \ -D taskmanager.memory.process.size=8192m \ -D taskmanager.numberOfTaskSlots=8 \ -D yarn.application.queue=flink \ -c com.example.demo.MyFlinkJob /opt/app/flink-demo.jar

这样做的另一个好处是不同作业互不影响。如果在flink-conf.yaml里写死并行度32，哪天提交一个本来只需4并行的小作业，也会强制分配32个并行，浪费大量资源。

6.4 状态后端和检查点目录的落地配置

生产环境的Flink on YARN一定要提前规划状态后端。我推荐的状态后端选择很直接：

• 有状态、需要精确一次语义的作业，用rocksdb，它把大状态溢出到本地磁盘，堆内存压力小。
• 状态较小，几十G以下的作业，用hashmap即可，最简单的内存状态后端。

对应的flink-conf.yaml配置：

state.backend: rocksdb state.checkpoints.dir: hdfs://node01:8020/flink/checkpoints state.savepoints.dir: hdfs://node01:8020/flink/savepoints execution.checkpointing.interval: 60s execution.checkpointing.min-pause: 30s

检查点目录一定放在HDFS这种共享存储上。YARN容器重启、TaskManager换节点之后，新启动的任务才能从最近一次检查点恢复，状态不丢。

如果RocksDB所需内存占比较高，可以调大托管内存：

taskmanager.memory.managed.fraction: 0.5

默认0.4，对RocksDB场景调到0.5能减少因堆外内存不足触发的状态写盘失败。

6.5 队列资源不足时先看哪几个指标

Flink作业提交后状态一直是ACCEPTED，既不启动也不报错，十有八九是YARN队列资源不够。按顺序查：

• yarn queue -status flink：看队列里当前使用量和待调度资源。
• yarn cluster -metrics：看集群整体可用资源。
• yarn application -status：看应用Diagnostics字段有没有“Waiting for available resources”之类的提示。

如果是队列资源满，可以选择等资源释放，也可以临时调整队列容量比例，但不建议直接调大yarn.scheduler.maximum-allocation-mb——这个参数控制单容器最大内存，调太高会让一个大容器独占队列资源，其他任务都进不来。

7. 高频故障复盘：容器反复重启、类冲突、日志定位的三类实际问题

7.1 TaskManager容器反复被杀：先怀疑内存再怀疑资源

现象很典型：作业提交后，Flink Web UI里能看到TaskManager在注册，但过一会儿全部消失，YARN上容器状态是KILLED或者COMPLETED，日志里出现Container killed by YARN for exceeding memory limits。

先不用急着改代码，直接按这个顺序查：

1. 看Flink Web UI的TaskManager日志，确认是不是OutOfMemoryError，以及是哪一种。
2. 如果没有OOM，看NodeManager日志，是不是物理内存超限。
3. 看YARN的诊断信息里，容器申请的内存大小和队列最高可用单容器内存是否匹配。

定位后调整策略：

• OOM堆内存：调大taskmanager.memory.process.size，同时确认托管内存占比。
• 物理内存超限：调大taskmanager.memory.jvm-overhead.fraction到0.15，必要时检查RocksDB内存占比。
• 资源不匹配：降低TM内存或slot数，保证单个容器申请不超过yarn.scheduler.maximum-allocation-mb。

7.2 类找不到或方法找不到：九成是依赖冲突

Flink on YARN环境下类冲突的排查比普通Java项目更绕，因为Flink进程的类加载器会分成几个层次：Flink框架自身类、Hadoop客户端类、用户作业jar包类。如果用户jar包里带了一份和Flink内部冲突的依赖，实际加载到哪个类完全取决于加载顺序。

我遇到最多的是提交HBase或HDFS相关扩展任务时，出现NoSuchMethodError或者ClassNotFoundException。排查思路：

mvn dependency:tree

先看用户jar里的依赖树，重点关注是否引入了和Flink内置版本不一致的hadoop-*、guava、netty、protobuf这些常见冲突源。

然后看Flink Web UI里该作业的Classloader信息，确认冲突的类是从哪个jar包加载的。

修复方式一般是两个：

1. 用户作业构建时用maven shade插件，把常用依赖relocate到自己的命名空间。
2. 把作业jar里和Flink核心相关的依赖设为provided，不要打进fat jar。

pom里Flink依赖我的基准写法：

<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency>

provided意味着Flink运行时用自己的版本，作业jar里不打包，这是减少类冲突最有效的一招。

7.3 状态一直是ACCEPTED但跑不起来：先看队列再看单容器上限

前面第6章提到过资源不足导致排队的情况，还有一个常见原因是单容器内存配置超过了YARN集群的单容器上限。YARN默认yarn.scheduler.maximum-allocation-mb可能是8G或者16G，如果你在Flink配置里写了taskmanager.memory.process.size: 16384m，单个容器内存超过上限，YARN会认为这个请求无法满足，作业就永远在排队。

怎么确认？看YARN诊断信息，如果出现类似Could not allocate container或者Invalid resource request，基本就是这个问题。

解决方式：

yarn application -status application_xxxx | grep Diagnostics

看到明确的资源请求报错后，把TM内存调小，或者把YARN的maximum-allocation-mb调大，二选一。我建议先调小Flink侧的内存，因为盲目调大全局单容器上限会破坏调度平衡。

7.4 日志聚合配置：没配置连日志都找不到

Flink on YARN排障最痛苦的不是问题难，而是日志拿不到。YARN默认yarn.log-aggregation-enable可能没开，任务一结束，容器日志直接清理，你想分析也找不到数据源。

在yarn-site.xml中开启日志聚合：

<property> <name>yarn.log-aggregation-enable</name> <value>true</value> </property> <property> <name>yarn.log-aggregation.retain-seconds</name> <value>604800</value> </property>

开启后，作业日志会被聚合到HDFS上，再用：

yarn logs -applicationId application_xxxx

就能拿到完整日志。这个配置强烈建议在Flink接入前就打开，不然后续所有故障排查都是盲人摸象。

Application模式下还有个细节，main方法本身是在AM容器里执行的。如果作业提交后YARN应用启动失败，优先查看AM日志而非TaskManager日志。AM日志通常会直接告诉你用户代码初始化阶段在哪一步崩了，比对着TaskManager日志猜Agent快得多。

这套Flink on YARN集群落地之后，我把排障思路固定在三个方向：先查YARN容器状态和日志，再查Flink内存模型配置，最后查用户依赖和队列资源。绝大多数问题都能在这三个方向里找到答案，真正需要深入作业内部逻辑的情况反而很少。希望这篇实战记录，能让你少走几次弯路。