我最近在给一套已有的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是改动最小、见效最快的方案。维度Flink StandaloneFlink on YARN资源来源独立物理/虚拟机器动态申请YARN容器多租户隔离需自行实现复用YARN队列故障恢复依赖外部脚本YARN容器级重启运维成本需要额外维护一套进程体系跟随YARN统一运维适用场景小规模验证、无现成YARN环境已有Hadoop集群的生产环境2. 版本与基础环境规划先定Flink和Hadoop的兼容连锁清单2.1 版本组合是怎么定下来的版本选型是Flink on YARN搭建中最容易忽略、又最容易踩爆的一环。Flink对Hadoop客户端版本并不是“任何版本都能随便配”。我这次遇到的情况是线上为Hadoop 3.3.x综合考虑后选了Flink 1.18JDK用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 规划一个合理的示例集群我做资源规划时按下面这个最小规模来打样三台机器足够跑通整套流程主机角色内存CPU核数说明node01HDFS NameNode / YARN ResourceManager / Flink客户端32G16提交作业的主入口node02HDFS DataNode / YARN NodeManager64G32运行Flink TaskManager容器node03HDFS DataNode / YARN NodeManager64G32运行Flink TaskManager容器这套规划下YARN集群的实际可用资源大约从node02和node03出node01主要承担管理角色。如果你机器多继续增加DataNode/NodeManager节点即可Flink不需要单独预留机器。2.4 基础环境四件套别偷懒正式搭建前花十分钟检查四件事所有节点时间同步。YARN对容器心跳有时间判断时钟偏差过大会出现节点被ResourceManager误判为异常进而导致容器反复调度。主机名和DNS解析。/etc/hosts要写好映射Flink提交时HDFS地址如果用主机名解析不了会直接失败。防火墙。至少保证ResourceManager Web端口、Flink Web UI所在端口、节点间通信端口不挡。客户端到YARN的提交权限。如果你用普通用户提交需要提前在YARN上给这个用户或用户组授权。3. 三种运行模式的差异拆解Session、Per-Job、Application如何取舍3.1 Session模式常驻会话适合临时查询Session模式启动一个长期驻留的YARN会话一个JobManager放在那里后面提交多个作业都复用它。好处是提交快、不需要每次启动新JobManager非常适合开发调试和临时跑一些小的分析作业。但Session模式的问题也出在“复用”上。所有任务共享同一个JobManager的资源与内存任务多了JobManager会成为瓶颈业务上如果有一个作业把资源占满其他作业只能排队。而且Session本身是一个常驻进程不及时关闭就会一直占用YARN队列资源。3.2 Per-Job模式一个作业一个JobManagerPer-Job模式是Flink 1.15之前生产环境最常用的模式。每次提交都会在YARN上启动一个独立的JobManager作业结束整个集群资源随之释放互不干扰。隔离性比Session好很多适合长周期的正式作业。不过Per-Job模式在提交过程中依赖客户端作为“中间人”。客户端提交作业后还要负责和JobManager通信客户端一旦退出作业有中断风险。这一点在Flink 1.15之后的版本里逐渐被弱化新版本更推荐用Application模式。3.3 Application模式把main方法也交给YARNApplication模式从Flink 1.11开始引入核心变化是用户的main方法在YARN的ApplicationMaster容器里执行。你本地只要提交整个作业jar包YARN拿到之后会在集群内启动AMAM直接跑main方法创建JobManager和后续的TaskManager。这个模式解决了Per-Job依赖客户端的问题。提交命令执行完毕本地终端就可以关了作业照样在YARN上跑。生产环境里长周期作业我基本默认用Application模式。3.4 我的选型建议使用场景推荐模式原因开发调试、临时SQLSession启动快、提交快生产环境独立作业Application生命周期完整、不依赖客户端多个紧密协作的小作业Session共享JobManager减少开销大规模正式集群的默认选择Application隔离性好、运维简单一句话总结如果你在规划新集群不用犹豫默认Application模式。Session只留给真正有“频繁提交短任务”诉求的场景。4. 从零开始的搭建步骤依赖环境、配置文件到首次提交4.1 先确认YARN集群有足够资源动手配置Flink之前我先在客户端验证Hadoop集群状态hdfs dfs -ls / yarn node -list -all yarn cluster -metricsyarn node -list -all能看到所有节点是否Health状态yarn cluster -metrics能看到集群总内存、可用内存、剩余核数。只有可用内存大于你即将申请的资源作业才能顺利跑起来。这一步别跳过节省排查时间的效果非常明显。我有一次配置完全没问题但作业一直ACCEPTED不启动查到最后是集群可用内存被其他任务占了YARN在排队等资源。4.2 解压Flink并配置flink-conf.yamlFlink目录结构很简单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-11env.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.jarsbin目录下的示例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 BufferTaskManager各slot之间传输数据需要的内存。JVM OverheadMetaspace、线程栈、JVM自身开销。如果只把堆内存调大堆外得不到保障任务处理高峰一来就会OOM容器被YARN杀掉然后不断重启看起来像集群不稳定实际是内存配比不合理。所以新版本的Flink强烈建议统一使用taskmanager.memory.process.size把整个进程的总内存额度交给Flink自己去拆分。这样最容易保证各部分均衡也最容易通过启动校验。5.2 一个具体的资源计算例子假设你有一个作业业务压力估算之后确定并行度需要16。按前面配置taskmanager.numberOfTaskSlots4那么需要的TaskManager数量就是TaskManager数量 16 / 4 4如果每个TaskManager进程总内存是4GJobManager进程内存是2G那么需要向YARN申请的容器内存总量是容器内存总量 4 * 4G 2G 18G看起来数字不大但你得同时考虑所在YARN队列的容量。如果队列最大可用内存是16G这个作业就要排队等待资源释放。调调并行度、降降TM内存往往就调度上去了。5.3 容器内存超限的坑为什么会有物理内存超限YARN容器有内存限制进程实际占用的物理内存超过容器申请值NodeManager会直接杀掉容器。Flink的进程总内存已经包含了JVM Overhead按理说不会超但如果某个TaskManager使用RocksDB状态后端且managed memory分配的比例不够RocksDB持续写磁盘前的缓存就会把堆外内存顶破。常见的三个直接原因taskmanager.memory.jvm-overhead.fraction默认0.1对某些状态应用来说偏小我一般调到0.15。并发slot太多每个slot上的子任务同时创建大量临时对象。用户代码里缓存了大的静态数据结构比如把几十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 nameyarn.scheduler.capacity.root.queues/name valuedefault,flink/value /property property nameyarn.scheduler.capacity.root.flink.capacity/name value40/value /property property nameyarn.scheduler.capacity.root.default.capacity/name value60/value /property这样根队列下面就有了default和flink两个子队列Flink专用队列占总资源40%。改完配置要执行yarn rmadmin -refreshQueues才能生效。6.2 提交作业时指定队列队列设置好之后提交作业时用-D动态指定./bin/flink run-application -t yarn-application \ -D yarn.application.queueflink \ -c com.example.demo.MyFlinkJob /opt/app/flink-demo.jar如果你希望某个集群所有Flink作业都默认走某队列可以在flink-conf.yaml里固定yarn.application.queue: flink6.3 运行参数最好用-D在提交时覆盖配置文件是默认值具体作业最好在执行命令里用-D覆盖。比如某个作业数据量大需要更多内存./bin/flink run-application -t yarn-application \ -D parallelism.default32 \ -D taskmanager.memory.process.size8192m \ -D taskmanager.numberOfTaskSlots8 \ -D yarn.application.queueflink \ -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。先不用急着改代码直接按这个顺序查看Flink Web UI的TaskManager日志确认是不是OutOfMemoryError以及是哪一种。如果没有OOM看NodeManager日志是不是物理内存超限。看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包加载的。修复方式一般是两个用户作业构建时用maven shade插件把常用依赖relocate到自己的命名空间。把作业jar里和Flink核心相关的依赖设为provided不要打进fat jar。pom里Flink依赖我的基准写法dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version${flink.version}/version scopeprovided/scope /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-clients/artifactId version${flink.version}/version scopeprovided/scope /dependencyprovided意味着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 nameyarn.log-aggregation-enable/name valuetrue/value /property property nameyarn.log-aggregation.retain-seconds/name value604800/value /property开启后作业日志会被聚合到HDFS上再用yarn logs -applicationId application_xxxx就能拿到完整日志。这个配置强烈建议在Flink接入前就打开不然后续所有故障排查都是盲人摸象。Application模式下还有个细节main方法本身是在AM容器里执行的。如果作业提交后YARN应用启动失败优先查看AM日志而非TaskManager日志。AM日志通常会直接告诉你用户代码初始化阶段在哪一步崩了比对着TaskManager日志猜Agent快得多。这套Flink on YARN集群落地之后我把排障思路固定在三个方向先查YARN容器状态和日志再查Flink内存模型配置最后查用户依赖和队列资源。绝大多数问题都能在这三个方向里找到答案真正需要深入作业内部逻辑的情况反而很少。希望这篇实战记录能让你少走几次弯路。