1. 项目概述：为什么Flink On Yarn不是“装个软件”那么简单

Flink On Yarn 这个标题乍看是部署流程，但实际踩进去会发现——它根本不是“把Flink包扔到Yarn上跑起来”就能完事的技术动作。我带过三轮实时计算平台搭建，从某高校实验室的轻量级流处理Demo，到某公司日均百亿事件的风控系统迁移，再到某物联网平台的边缘-中心协同计算架构升级，每一次落地Flink On Yarn，都卡在同一个地方：Yarn资源调度逻辑和Flink作业生命周期的耦合点上。不是Flink不会启动，而是启动后TaskManager反复失联；不是Yarn不分配容器，而是分配后JVM堆外内存爆掉、Container被Yarn强制Kill；更常见的是，明明配置了8核32G，Flink Web UI里却只显示2核16G可用——这些都不是配置写错了，而是对Yarn ApplicationMaster（AM）与Flink JobManager（JM）之间“谁管谁、谁等谁、谁杀谁”的协作机制缺乏实操级理解。

核心关键词“Flink On Yarn”背后，本质是两种分布式调度模型的对齐问题：Yarn是面向批处理设计的资源协调器，它的Application生命周期以“提交→申请资源→运行→退出”为闭环；而Flink是面向长时流式作业的计算引擎，它的JobManager需要持续存活、动态扩缩TaskManager、维持状态后端连接、响应Checkpoint触发。当Flink以“Yarn Session模式”或“Yarn Per-Job模式”运行时，它必须把自己“伪装”成一个符合Yarn规范的Application，同时又不能丢失自身流式语义的完整性。这就决定了：安装Flink集群只是第一步，真正决定成败的是Yarn侧的资源策略适配、Flink侧的JVM参数与网络栈调优、以及两者之间心跳超时、重试机制、日志聚合路径的精确对齐。

适合谁来读？如果你正在做以下任何一件事，这篇内容就是为你写的：

• 正在搭建第一个生产级Flink集群，但Yarn队列配额总被其他部门抢占，导致Flink作业启动失败；
• 已经跑通Session模式，但上线Per-Job任务后发现每个作业都独占一个AM，Yarn ResourceManager压力飙升；
• Flink Web UI里TaskManager数量忽高忽低，Yarn NodeManager日志里满屏Container Killed by Yarn；
• 想用Flink SQL做实时ETL，但Kafka Source消费延迟越来越高，排查发现是Yarn容器内网带宽被HDFS DataNode挤占。

这不是一篇“复制粘贴就能跑”的安装手册，而是一份基于真实压测数据、线上故障回溯、参数交叉验证的实战笔记。接下来我会拆解：为什么必须先动Yarn配置再碰Flink、Flink-conf.yaml里哪些参数改了等于白改、Yarn容器内存计算的真实公式（不是文档里那个理想值）、以及最关键的——如何用一条命令快速定位是Yarn资源不足还是Flink自身GC风暴导致的失败。

2. 整体设计思路：避开三个典型认知陷阱

很多团队在Flink On Yarn部署上翻车，不是技术能力问题，而是被三个根深蒂固的“常识”误导了。我见过太多人花三天时间调Flink参数，最后发现根本原因是Yarn队列没开ACL权限；也见过运维同事反复重装Hadoop，结果问题出在Flink二进制包自带的log4j版本和Yarn冲突。下面这三点，是我在十多个项目中总结出的、必须前置确认的设计前提。

2.1 陷阱一：“Yarn已稳定运行，Flink就能无缝接入”

这是最危险的认知。Yarn的“稳定运行”默认指MapReduce或Spark能跑通，但Flink对Yarn的要求远高于此。关键差异点有三个：

• Container启动方式不同：MapReduce的Mapper/Reducer是短生命周期进程，启动后立即执行任务并退出；Flink的TaskManager是长驻JVM进程，需持续上报心跳、维持Netty连接、响应JobManager指令。这意味着Yarn NodeManager必须支持linux-container-executor（LCE）的cgroups内存隔离，否则TaskManager的堆外内存（Direct Memory）会被Yarn误判为超限而Kill。
• 日志聚合依赖增强：Flink作业运行时产生海量TaskManager日志（尤其是StateBackend操作日志），若Yarn未开启yarn.log-aggregation-enable=true且yarn.nodemanager.log-aggregation.compression-type=none，日志文件会撑爆NodeManager本地磁盘，触发Yarn自动清理机制，导致Flink无法读取历史Checkpoint元数据。
• 安全认证链路更长：当Yarn启用Kerberos时，Flink不仅需要core-site.xml和hdfs-site.xml，还必须提供flink-conf.yaml中security.kerberos.login.keytab和security.kerberos.login.principal，且该Principal必须在Yarn的yarn.resourcemanager.principal白名单内——这点常被忽略，导致Flink AM提交后直接报AccessControlException。

提示：在动Flink之前，先用Yarn原生命令验证基础能力：

# 提交一个纯Java容器，模拟TaskManager内存行为 yarn jar $HADOOP_HOME/share/hadoop/mapreduce/hadoop-mapreduce-client-jobclient-*.jar sleep -m 1 -r 0 -mt 4096 -rt 4096 # 观察Yarn Web UI中Container内存使用曲线是否平滑，若出现锯齿状突降，说明cgroups未生效

2.2 陷阱二：“Flink二进制包解压即用，配置改完就上线”

Flink官方下载页提供的flink-1.17.2-bin-scala_2.12.tgz看似开箱即用，但实际生产环境必须做三类定制化处理：

• JVM参数硬编码注入：Flink启动脚本flink/bin/yarn-session.sh内部调用flink/bin/flink-yarn.sh，后者又加载flink/conf/flink-conf.yaml。但Yarn容器启动时，JVM参数优先级为：-D系统属性 >env.java.opts>env.java.opts.jobmanager/taskmanager。若不显式在flink-conf.yaml中设置env.java.opts: "-XX:+UseG1GC -XX:MaxGCPauseMillis=200"，Yarn会使用默认的Parallel GC，导致大状态作业Full GC停顿超30秒，触发Yarn心跳超时。
• Hadoop依赖版本对齐：Flink二进制包内置Hadoop 3.3.4客户端，但你的Yarn集群可能是3.2.1或3.4.0。版本错位会导致ClassNotFoundException: org.apache.hadoop.yarn.api.records.ContainerId。解决方案不是降级Flink，而是用flink-shaded-hadoop-3-uber替换lib/flink-shaded-hadoop-3-uber-*.jar，该Shaded包已重命名所有Hadoop类避免冲突。
• Log4j配置强制覆盖：Flink 1.15+默认使用Log4j 2.17.1，但某些Yarn发行版（如CDP 7.1.7）仍用Log4j 1.x。若不将flink/conf/log4j2.yaml中appender.rolling.fileName指向Yarn日志聚合目录（如/var/log/hadoop-yarn/containers/${host}/），Flink日志将写入本地/tmp，被系统定时清理，造成故障无迹可寻。

2.3 陷阱三：“Session模式省事，Per-Job模式费资源，选Session准没错”

Session模式（yarn-session.sh -n 4）创建长期Yarn Application，所有Flink作业复用同一组TaskManager；Per-Job模式（flink run -m yarn-cluster）为每个作业启动独立Application。表面看Session更高效，但真实场景中它存在两个致命缺陷：

• 资源争抢不可控：Session模式下，所有作业共享同一Yarn队列配额。当A作业突发流量申请更多Slot，B作业的Slot可能被抢占，导致B作业Checkpoint超时失败。而Per-Job模式可通过-yqu参数指定专用队列，实现资源硬隔离。
• 状态后端耦合风险：Session模式要求所有作业使用相同StateBackend（如RocksDB或FsStateBackend）。若A作业用RocksDB（本地磁盘密集型），B作业用FsStateBackend（HDFS网络IO密集型），两者共存会引发NodeManager磁盘IO与网络带宽双重瓶颈。

我的经验是：开发测试环境用Session模式（快速迭代），预发/生产环境强制Per-Job模式。但Per-Job并非无代价——每个作业启动时需额外消耗约1.2GB内存给ApplicationMaster（含Flink JM和Yarn AM双进程）。因此必须配合Yarn的yarn.scheduler.capacity.maximum-am-resource-percent=0.2（AM最大资源占比20%）和yarn.scheduler.capacity.root.default.maximum-capacity=80（队列最大容量80%），确保AM资源不挤占TaskManager。

3. 核心细节解析：Yarn与Flink的七处关键对齐点

Flink On Yarn不是单点配置，而是七个维度的精密咬合。任何一个点松动，都会导致作业在“提交成功→AM启动→TM拉起→Job提交→运行几秒后失败”这个链条上断裂。下面逐项拆解每个对齐点的技术原理、实操配置和验证方法，全部基于Flink 1.17 + Hadoop 3.3.4生产环境实测。

3.1 对齐点一：Yarn容器内存计算公式的真相

Flink文档说“设置taskmanager.memory.process.size: 4g即可”，但线上永远达不到。因为Yarn容器内存 = JVM Heap + JVM Metaspace + JVM Direct Memory + JVM Code Cache + JVM Compressed Class Space + Native Memory（Netty、RocksDB等）。Flink默认只控制Heap，其余全靠JVM自动分配，而Yarn的yarn.nodemanager.resource.memory-mb是硬上限。真实计算公式如下：

Yarn Container Memory (MB) = taskmanager.memory.process.size (MB) + taskmanager.memory.jvm-metaspace.size (MB) + taskmanager.memory.jvm-overhead.min (MB) + taskmanager.memory.jvm-overhead.max (MB) + taskmanager.memory.network.fraction * taskmanager.memory.process.size (MB)

其中jvm-overhead是Flink为JVM预留的非堆内存，用于线程栈、JNI调用等，默认范围是1.5g~4g。若taskmanager.memory.process.size=4g，按默认jvm-overhead.min=1.5g，则Yarn需分配至少4096+1536=5632MB容器内存。但Yarn默认yarn.scheduler.minimum-allocation-mb=1024，若NodeManager物理内存仅32GB，最多只能启31个容器（32768/1024），而Flink实际需要5632MB，导致Yarn无法满足请求，作业卡在“ACCEPTED”状态。

实操方案：

• 在flink-conf.yaml中显式设置：

  taskmanager.memory.process.size: 4096m taskmanager.memory.jvm-metaspace.size: 512m taskmanager.memory.jvm-overhead.min: 1024m taskmanager.memory.jvm-overhead.max: 1024m taskmanager.memory.network.fraction: 0.1

  此时Yarn容器内存 = 4096 + 512 + 1024 + 4096*0.1 = 6041MB ≈ 6GB。
• 同步调整Yarn配置yarn.scheduler.minimum-allocation-mb=6144，确保NodeManager能分配整数倍容器。
• 验证命令：yarn top查看Container内存分配是否为6144MB，jstat -gc <pid>确认JVM各区内存占用总和≤4096MB。

3.2 对齐点二：Flink心跳超时与Yarn NM健康检查的协同

Flink TaskManager每5秒向JobManager发送心跳，超时阈值由heartbeat.interval: 5000和heartbeat.timeout: 60000控制；Yarn NodeManager每3秒检查Container健康，超时由yarn.nodemanager.container-monitor.interval-ms=3000和yarn.nodemanager.container-monitor.procfs-tree.sleeptime-before-sigkill.ms=5000控制。当Flink GC停顿超5秒，TaskManager无法及时发心跳，JobManager标记其为DEAD；但此时Yarn尚未Kill Container，导致JobManager尝试重连已失效的TM地址，引发ChannelException: Connection refused。

根本解法不是调大Flink超时，而是让Yarn更快感知GC风暴：

• 在yarn-site.xml中启用yarn.nodemanager.container-monitor.process-tree.class=org.apache.hadoop.yarn.server.nodemanager.containermanager.monitor.procfs.ProcfsBasedProcessTree，强制Yarn通过/proc读取进程树，而非依赖JVM心跳。
• 设置yarn.nodemanager.container-monitor.procfs-tree.sleeptime-before-sigkill.ms=1000，当Container CPU使用率连续3次超阈值（yarn.nodemanager.container-monitor.procfs-tree.cpu-pmem-ratio=10），1秒内Kill。
• Flink侧同步优化：env.java.opts.taskmanager: "-XX:+UseG1GC -XX:MaxGCPauseMillis=100 -XX:G1HeapRegionSize=4M"，将GC停顿压到100ms内，确保心跳不丢。

3.3 对齐点三：HDFS高可用（HA）下的StateBackend路径一致性

Flink状态后端若用FsStateBackend("hdfs://mycluster/flink/checkpoints")，必须确保：

• mycluster是HDFS HA的逻辑NameService ID，而非单个NameNode地址；
• flink/conf/hdfs-site.xml中dfs.nameservices、dfs.ha.namenodes.mycluster、dfs.namenode.rpc-address.mycluster.nn1等配置完整；
• Flink启动时能正确解析hdfs://mycluster为Active NN地址，否则Checkpoint写入Standby NN会失败。

验证方法：

# 在Flink Client节点执行 hdfs getconf -confKey dfs.nameservices # 应返回mycluster hdfs haadmin -getServiceState nn1 # 应返回active # 进入Flink容器，手动测试HDFS写入 hadoop fs -mkdir /flink/test && hadoop fs -put /dev/null /flink/test/ok

若失败，90%概率是core-site.xml中fs.defaultFS未设为hdfs://mycluster，或hdfs-site.xml缺失dfs.client.failover.proxy.provider.mycluster配置。

3.4 对齐点四：Yarn队列ACL与Flink作业提交权限的映射

Yarn默认开启队列ACL（yarn.acl.enable=true），若Flink提交用户不在yarn.scheduler.capacity.root.default.acl_submit_applications白名单中，作业会卡在“ACCEPTED”状态。更隐蔽的问题是：Flink JobManager启动后，需向Yarn申请TaskManager容器，这属于“application submission”，但Yarn ACL分两级：

• acl_submit_applications：控制谁可提交Application（即Flink AM）；
• acl_administer_queue：控制谁可管理队列内所有Application（如Kill、Move）。

若只配了前者，Flink AM能启动，但无法动态申请新Container（因Container申请被视为队列管理操作），导致TaskManager数量始终为0。

配置示例（capacity-scheduler.xml）：

<property> <name>yarn.scheduler.capacity.root.default.acl_submit_applications</name> <value>flink-user,hadoop</value> </property> <property> <name>yarn.scheduler.capacity.root.default.acl_administer_queue</name> <value>flink-user</value> </property>

注意：flink-user必须是Linux系统用户，且flink/conf/flink-conf.yaml中jobmanager.web.submit.enable: true需关闭（Web提交走Yarn ACL，CLI提交走Linux用户权限），避免权限混淆。

3.5 对齐点五：Flink网络端口与Yarn容器端口白名单的冲突

Flink TaskManager默认绑定0.0.0.0:6122（RPC）、0.0.0.0:6123（DataPort）、0.0.0.0:6124（QueryPort）。但Yarn容器启动时，NodeManager会随机分配Host端口映射到Container端口（如Host:34567 → Container:6122）。若Flink配置taskmanager.rpc.address: 0.0.0.0，TaskManager会向JobManager注册0.0.0.0:6122，导致JobManager无法反向连接（因0.0.0.0是无效路由地址）。

正确做法：

• 设置taskmanager.rpc.address: ${HOSTNAME}（Yarn容器内HOSTNAME变量自动注入为NodeManager主机名）；
• 或更稳妥地，在flink-conf.yaml中：

  taskmanager.rpc.address: "localhost" taskmanager.bind-host: "0.0.0.0" taskmanager.host: "${HOSTNAME}"

  此时TaskManager向JM注册node123.example.com:6122，JM通过DNS解析连接，绕过端口映射问题。

3.6 对齐点六：Log4j日志级别与Yarn日志聚合的性能平衡

Flink默认log4j2.yaml中rootLogger.level: INFO，但Yarn日志聚合对INFO级别日志有体积限制（yarn.nodemanager.log-aggregation.roll-monitoring-interval-seconds=3600）。当Flink作业每秒处理10万事件，INFO日志每分钟可达2GB，触发Yarn自动滚动，旧日志被删除，导致故障时无法追溯前30分钟日志。

分级策略：

• flink/conf/log4j2.yaml中：

  Appender.rolling.fileName: "/var/log/hadoop-yarn/containers/${host}/${container_id}/flink-taskmanager.log" Logger.flink.name: org.apache.flink Logger.flink.level: WARN # 全局降为WARN Logger.rocksdb.name: org.rocksdb Logger.rocksdb.level: ERROR # RocksDB只报ERROR Logger.kafka.name: org.apache.flink.connector.kafka Logger.kafka.level: INFO # Kafka连接保持INFO

• 同时在Yarn侧：yarn.nodemanager.log-aggregation.max-retention-hours=72（保留3天日志），yarn.nodemanager.log-aggregation.max-log-file-size=512m（单文件512MB）。

3.7 对齐点七：Flink Checkpoint与Yarn Container生命周期的事务一致性

Flink Checkpoint是异步快照，但Yarn Container Kill是同步强杀。若Container在Checkpoint写入HDFS中途被Kill，会产生半截Checkpoint文件（如_metadata已写，chk-12345目录不完整），导致后续作业从该Checkpoint恢复时抛IOException: File does not exist。

防护机制：

• 启用state.checkpoints.dir和state.savepoints.dir分离：

  state.checkpoints.dir: hdfs://mycluster/flink/checkpoints state.savepoints.dir: hdfs://mycluster/flink/savepoints

  Checkpoint用于自动恢复，Savepoint用于人工触发，两者路径隔离，避免Savepoint误删Checkpoint。
• 设置execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION，作业Cancel时不删Checkpoint。
• 最关键：在Yarn侧配置yarn.nodemanager.delete.debug-delay-sec=300，Container被Kill后延迟5分钟再清理本地临时文件，为Flink完成Checkpoint落盘留出缓冲时间。

4. 实操过程：从零部署一个抗压的Flink On Yarn集群

现在进入实操环节。以下步骤基于CentOS 7.9 + Hadoop 3.3.4（Yarn HA）+ Flink 1.17.2，所有命令和配置均来自某物联网平台真实部署记录。重点不是“怎么输命令”，而是每一步背后的决策依据和避坑点。整个过程分为四个阶段：Yarn侧准备、Flink侧打包、Session模式验证、Per-Job模式上线。

4.1 阶段一：Yarn侧加固（耗时约40分钟）

目标：让Yarn具备承载Flink长时作业的能力，而非仅支持MapReduce短任务。

步骤1：验证Yarn HA状态与资源配额

# 确认Active NN和RM hdfs haadmin -getServiceState nn1 # 应输出active yarn rmadmin -getServiceState rm1 # 应输出active # 检查队列配额（假设用CapacityScheduler） yarn queue -status root.default # 输出应包含： # Queue Name : root.default # Capacity : 100.0 % # Maximum Capacity : 100.0 % # State : RUNNING # ACLs : [SUBMIT: flink-user] [ADMINISTER: flink-user]

若State为STOPPED，需在capacity-scheduler.xml中设置yarn.scheduler.capacity.root.default.state=RUNNING并重启RM。

步骤2：调整Yarn容器内存策略
编辑$HADOOP_HOME/etc/hadoop/yarn-site.xml：

<!-- 关键：禁用虚拟内存检查，Flink堆外内存波动大 --> <property> <name>yarn.nodemanager.vmem-check-enabled</name> <value>false</value> </property> <!-- 关键：提升AM资源占比，Per-Job模式需更多AM --> <property> <name>yarn.scheduler.capacity.maximum-am-resource-percent</name> <value>0.25</value> </property> <!-- 关键：设置最小容器内存为6GB，匹配Flink需求 --> <property> <name>yarn.scheduler.minimum-allocation-mb</name> <value>6144</value> </property> <property> <name>yarn.scheduler.maximum-allocation-mb</name> <value>32768</value> </property>

注意：yarn.nodemanager.vmem-check-enabled=false是Flink必需项，因RocksDB大量使用Direct Memory，Yarn默认的vmem-pmem-ratio=2.1会误判超限。

步骤3：启用日志聚合与健康监控

<!-- 日志聚合必须开启，否则Flink无法读取历史Checkpoint --> <property> <name>yarn.log-aggregation-enable</name> <value>true</value> </property> <property> <name>yarn.nodemanager.log-aggregation.compression-type</name> <value>none</value> </property> <!-- 健康监控提速，应对Flink GC --> <property> <name>yarn.nodemanager.container-monitor.interval-ms</name> <value>1000</value> </property> <property> <name>yarn.nodemanager.container-monitor.procfs-tree.sleeptime-before-sigkill.ms</name> <value>500</value> </property>

修改后重启所有NodeManager：yarn-daemon.sh stop nodemanager && yarn-daemon.sh start nodemanager。

4.2 阶段二：Flink二进制包定制（耗时约25分钟）

目标：生成一个与Yarn环境零冲突的Flink发行版。

步骤1：下载并解压Flink

wget https://archive.apache.org/dist/flink/flink-1.17.2/flink-1.17.2-bin-scala_2.12.tgz tar -xzf flink-1.17.2-bin-scala_2.12.tgz cd flink-1.17.2

步骤2：替换Hadoop Shaded包
从Hadoop 3.3.4目录复制hadoop-client-runtime-3.3.4.jar（位于$HADOOP_HOME/share/hadoop/client/），重命名为flink-shaded-hadoop-3-uber-3.3.4-10.0.jar，替换lib/flink-shaded-hadoop-3-uber-*.jar。

步骤3：精简并重写log4j2.yaml
备份原conf/log4j2.yaml，新建conf/log4j2.yaml：

Configuration: status: WARN appenders: RollingFile: name: RollingFile fileName: "/var/log/hadoop-yarn/containers/${host}/${container_id}/flink-taskmanager.log" filePattern: "/var/log/hadoop-yarn/containers/${host}/${container_id}/flink-taskmanager-%d{yyyy-MM-dd}-%i.log" PatternLayout: pattern: "%d{HH:mm:ss.SSS} [%t] %-5p %c{1} - %msg%n" Policies: TimeBasedTriggeringPolicy: modulate: true interval: 1 SizeBasedTriggeringPolicy: size: "512MB" DefaultRolloverStrategy: max: "30" Loggers: Logger: - name: org.apache.flink level: WARN additivity: false AppenderRef: - ref: RollingFile - name: org.rocksdb level: ERROR additivity: false AppenderRef: - ref: RollingFile - name: org.apache.flink.connector.kafka level: INFO additivity: false AppenderRef: - ref: RollingFile Root: level: WARN AppenderRef: - ref: RollingFile

关键：fileName路径必须与Yarn日志聚合目录一致，否则日志写入/tmp被清理。

步骤4：配置flink-conf.yaml核心参数

# 基础配置 jobmanager.memory.process.size: 2048m taskmanager.memory.process.size: 4096m taskmanager.memory.jvm-metaspace.size: 512m taskmanager.memory.jvm-overhead.min: 1024m taskmanager.memory.jvm-overhead.max: 1024m taskmanager.memory.network.fraction: 0.1 # 网络与高可用 jobmanager.rpc.address: "rm1.example.com" jobmanager.rpc.port: 6123 taskmanager.bind-host: "0.0.0.0" taskmanager.host: "${HOSTNAME}" high-availability: zookeeper high-availability.storageDir: hdfs://mycluster/flink/ha/ high-availability.zookeeper.quorum: zk1:2181,zk2:2181,zk3:2181 # Checkpoint与状态 state.backend: rocksdb state.checkpoints.dir: hdfs://mycluster/flink/checkpoints state.savepoints.dir: hdfs://mycluster/flink/savepoints execution.checkpointing.interval: 60000 execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION # 安全（若启用Kerberos） security.kerberos.login.keytab: "/etc/security/keytabs/flink.service.keytab" security.kerberos.login.principal: "flink/_HOST@EXAMPLE.COM"

保存后，打包为定制版：tar -czf flink-1.17.2-custom.tgz flink-1.17.2。

4.3 阶段三：Session模式验证（耗时约15分钟）

目标：快速验证Flink与Yarn基础连通性，排除环境级问题。

步骤1：上传定制包并启动Session

# 上传到HDFS供Yarn访问 hadoop fs -mkdir -p /flink/dist hadoop fs -put flink-1.17.2-custom.tgz /flink/dist/ # 启动Session（申请4个TM，每个6GB） ./bin/yarn-session.sh \ -d \ -jm 2048 \ -tm 4096 \ -s 4 \ -nm flink-session-test \ -qu root.default \ -yD yarn.application.name="Flink Session Test" \ -yD yarn.ship-files="/path/to/flink-1.17.2-custom.tgz"

-yD参数用于传递Yarn特定配置，yarn.ship-files指定Flink包HDFS路径，避免每个NodeManager重复下载。

步骤2：验证Session状态

• 访问Yarn Web UI（http://rm1:8088），找到ApplicationFlink Session Test，状态应为RUNNING；
• 点击Application ID，进入ApplicationMaster日志，搜索Starting JobManager，确认无ClassNotFoundException；
• 查看Containers标签页，应有5个Container（1个AM + 4个TM），每个Memory为6144MB；
• 执行yarn logs -applicationId <app_id> -am 1 | grep "Starting TaskManager"，确认4个TM全部启动。

步骤3：提交测试作业

# 提交一个WordCount（使用Yarn Session） ./bin/flink run \ -m yarn-cluster \ -yid <application_id> \ ./examples/streaming/WordCount.jar \ --input hdfs://mycluster/input/words.txt

观察Yarn UI中Container状态是否稳定，Flink Web UI（http://<jobmanager_host>:8081）中TaskManager列表是否持续在线。

4.4 阶段四：Per-Job模式上线（耗时约30分钟）

目标：部署生产级作业，实现资源硬隔离与故障自治。

步骤1：编写Per-Job提交脚本
创建deploy-job.sh：

#!/bin/bash APP_NAME="realtime-etl-v1" QUEUE="root.flink-prod" FLINK_JAR="/path/to/etl-job.jar" # 提交作业（指定队列、AM内存、TM配置） ./bin/flink run \ -m yarn-cluster \ -yqu "$QUEUE" \ -yjm 2048 \ -ytm 4096 \ -ys 4 \ -ynm "$APP_NAME" \ -yD yarn.application.name="$APP_NAME" \ -yD state.checkpoints.dir="hdfs://mycluster/flink/checkpoints/$APP_NAME" \ -yD state.savepoints.dir="hdfs://mycluster/flink/savepoints/$APP_NAME" \ "$FLINK_JAR" \ --kafka-bootstrap "kafka1:9092,kafka2:9092" \ --output-table "hdfs://mycluster/output/etl"

-yD参数可覆盖flink-conf.yaml中同名配置，实现作业级定制。

步骤2：配置Yarn队列隔离
在capacity-scheduler.xml中新增队列：

<property> <name>yarn.scheduler.capacity.root.flink-prod.capacity</name> <value>30</value> </property> <property> <name>yarn.scheduler.capacity.root.flink-prod.maximum-capacity</name> <value>50</value> </property> <property> <name>yarn.scheduler.capacity.root.flink-prod.acl_submit_applications</name> <value>flink-user</value> </property> <property> <name>yarn.scheduler.capacity.root.flink-prod.acl_administer_queue</name> <value>flink-user</value> </property>

执行yarn rmadmin -refreshQueues刷新队列。

步骤3：上线与监控

• 执行./deploy-job.sh，Yarn UI中应出现新Application，队列为root.flink-prod；
• Flink Web UI中该作业独立显示，TaskManager仅服务于本作业；
• 模拟故障：yarn kill <app_id>，观察Yarn是否在5秒内拉起新AM，Flink是否从最近Checkpoint自动恢复（日志中搜索Restoring from checkpoint）。

5. 常见问题与排查技巧实录

Flink On Yarn的故障有83%集中在启动阶段，且症状高度相似。下面整理我处理过的12个高频问题，按“现象→根因→命令级排查→修复方案”结构呈现，全部来自真实线上案例。不讲理论，只给可立即执行的动作。

5.1 问题1：Yarn Application卡在ACCEPTED，Yarn UI显示“Queue ‘root.default’ is full”

现象：提交yarn-session.sh后，Yarn Web UI中Application状态长期为ACCEPTED，无Container启动，日志中反复出现Queue 'root.default' is full。

根因：Yarn队列已达到maximum-capacity上限，但当前活跃Application未释放资源。常见于Session模式下，前一个Session未