1. 为什么生产环境我最终选了 Flink On Yarn做实时计算这几年Flink 的部署方式我基本都折腾过一遍。从最开始图省事搞 Standalone 集群到后来被线上事故教训得老老实实切到 Flink On Yarn中间踩过的坑够写一本小册子。今天这篇就专门聊聊 Flink On Yarn 这套部署方案好好说清楚它到底解决什么怎么选模式以及真正落地时那些配置和排查手段。1.1 独立集群模式的痛点资源是最大的问题先说 Standalone 模式。它的部署逻辑很简单固定启动 N 个 TaskManager每个 TaskManager 固定分配好内存和 slot 数然后所有作业都在这个池子里跑。听起来挺省事但一上生产就露馅。首先是资源预估值很难定。你开 20 个 TaskManager作业多了资源不够作业少了资源白白闲着晚上低谷期整个集群空转烧机器。我见过一个团队固定开 30 个 TaskManager结果凌晨低谷期利用率不到 10%白天高峰期又排队典型的钱花了没办好事。其次是作业之间的干扰完全无法隔离。Standalone 模式下所有作业共享 TaskManager 的内存和 CPU一个作业内存溢出往往把整个 TaskManager 拖垮同节点上的其他作业跟着陪葬。更麻烦的是 JobManager 只有一个集群级故障直接导致所有作业一起重启。还有一个隐形麻烦Standalone 集群本身需要单独运维。监控、扩缩容、版本升级全都得自己搞。如果你公司已经有 Hadoop 生态了那这套东西就相当于多维护了一个小集群成本高收益低。1.2 Yarn 调度的天然优势复用生态不做重复劳动Flink On Yarn 的核心逻辑很简单把 Flink 集群当成一个 Yarn 应用来提交由 Yarn 负责资源分配、容器启动和失败重启。这样做的好处非常实际。第一资源池是动态的作业提交时向 Yarn 申请资源作业结束后容器自动释放队列里的资源可以给其他任务用。白天跑实时作业晚上还能跑离线批处理集群利用率直线上升。第二每个作业默认拥有独立的 JobManager 和 TaskManager作业之间隔离性拉满一个作业炸了不影响别人。第三扩缩容不用手动加节点调整并行度或队列配置后重启作业即可Yarn 自动重新分配容器。更关键的是大多数公司的服务器上早就部署好了 Yarn 集群。有了这一层Flink 就不需要单独引入 Kubernetes 或自建资源管理平台直接对接现有的资源调度体系权限、队列、配额都能复用统一管控。所以说白了Flink On Yarn 就是”花一份运维成本吃两份生态红利“。它把实时计算任务的资源管理外包给 Yarn自己专注做计算逻辑这正是生产环境最需要的稳定性。2. Session 与 Per-Job两种模式到底怎么选Flink On Yarn 下还藏着两条岔路Yarn Session 模式和 Yarn Per-Job 模式。很多初学者在这里被绕晕其实这俩的概念和适合场景差异挺大选错了后面全是坑。2.1 Session 模式共享集群的便利与代价Session 模式的操作方式大致是先在 Yarn 上启动一个常驻的 Flink 集群这个集群的 JobManager 和一组 TaskManager 一直运行着之后所有作业都提交到这个共享集群里跑不再单独申请资源。它的最大优势是响应快。作业提交上去几秒钟就能跑起来省去了每次申请容器和拉起 JobManager 的过程。开发调试阶段特别舒服我经常一边改代码一边提交测试交互体验很像本地环境。但代价也很明显。所有作业共享同一个 JobManager 和同一批 TaskManager一旦某个作业出现内存溢出或异常 Full GC很容易拖慢整个集群严重时甚至导致所有作业集体重启。Session 模式下资源隔离基本为零生产环境用久了就变成“比谁运气好”的游戏。另外Session 集群的 JobManager 如果挂掉上面挂着的所有作业全部受影响哪怕其他作业啥事没有也得跟着一起重来。所以在线上严肃场景里Session 模式我基本只拿来应急或跑低频小任务。2.2 Per-Job 模式一个作业一个集群的“包间”Per-Job 模式则像一个一个独立的包间每提交一个作业Yarn 就单独拉起一套 Flink 集群包含自己的 JobManager 和专属 TaskManager作业跑完整个集群自动释放。这种模式的隔离性是最好的。每个作业独占容器资源内存溢出、Full GC、TaskManager 挂掉都只影响当前作业不会波及其他任务。而且 JobManager 也是独立的一个作业的调度器出问题不会干扰另一个作业。付出的代价是调度开销。每次提交作业都要重新走一遍 Yarn 资源申请、启动 JobManager、拉起 TaskManager 的流程启动时间通常在分钟级。对那种频繁启停的实验性作业来说这个等待确实让人心焦。但生产作业的特性决定了 Per-Job 才是主力。线上作业往往长期运行、需要稳定资源保证偶尔重启一次虽然有点慢但换来的稳定性和隔离性是完全值得的。我给团队定的规矩就是凡是跑正式业务的作业一律用 Per-Job只有开发联调或临时跑数才用 Session。2.3 选型决策表与注意事项选 Session 还是 Per-Job最终还是要结合自己的使用场景来判断。我整理了一张简表基本能覆盖大多数情况。对比维度Session 模式Per-Job 模式集群生命周期常驻需手动管理随作业启停自动释放作业启动耗时秒级分钟级资源隔离性差相互影响好互不干扰JobManager 故障影响影响全部作业只影响当前作业运维复杂度中需监控集群状态低作业即集群适用场景调试、小任务、交互式分析生产作业、长稳任务、重要业务有几个容易栽跟头的点提前说一下。第一Session 模式下如果队列资源不够所有共享作业会一起排队排查起来很头疼后来我一般单独给 Session 队列设置资源上限。第二Per-Job 模式每个作业都会申请自己的 JobManager如果同时提交大量作业Yarn 的 AM 数量会暴涨注意检查集群的 AM 配额。第三无论哪种模式都要确认 Yarn 的调度器类型Capacity Scheduler 队列配置不当会导致资源互相抢占。3. 动手实操环境准备与配置细节部署 Flink On Yarn 并不需要把 Flink 包装到每个节点核心思想是”客户端提交 Yarn 动态分配“。但环境准备和参数配置这一步做不好后面就是连环坑所以我把这部分拆开细讲。3.1 基础环境依赖缺一不可先说前置条件。Flink On Yarn 要求客户端机器能访问 Yarn 集群所以 Hadoop 客户端是必须的。最标准的做法是在提交作业的机器上安装 Hadoop 客户端并配好环境变量不推荐把整个 Hadoop 装到 Flink 节点上。需要确认的核心环境变量有这几个JAVA_HOMEFlink 脚本依赖、HADOOP_HOME提供 hadoop 命令、HADOOP_CONF_DIR指向 Yarn 和 HDFS 的配置目录。后面这个 HADOOP_CONF_DIR 特别关键Flink 提交作业时会自动读取里面的 yarn-site.xml、core-site.xml、hdfs-site.xml。还有一类问题经常被忽略Hadoop 原生库。如果集群开启了对 HDFS 的权限校验或者 Yarn 启用了 ACL客户端必须有对应的 kerberos 认证配置否则提交时直接秒报认证失败。我建议在正式跑业务前先手动执行一下 hdfs dfs -ls / 和 yarn node -list 这两个命令确认基础连通性没问题再继续。3.2 核心配置项逐个拆解环境就绪后就要面对 flink-conf.yaml这份文件里的参数直接决定作业能跑多快、会不会挂。很多新手一上来就乱抄网上的配置完全不知道每个参数是干什么的。下面挑几个影响最大的逐一说。先看 jobmanager.memory.process.size。这个值决定 JobManager 能用的总内存。Per-Job 模式下 JobManager 是独立容器内存给太小容易 OOM。我一般建议生产环境至少给 1-2GB取决于作业的状态规模和并发量。注意这个参数是进程总内存包含堆内和堆外官方推荐优先设置它而不是 jobmanager.memory.heap.size原因是为了让 Flink 自己处理堆内外分配减少人为调参头痛。taskmanager.memory.process.size 同样重要它决定每个 TaskManager 容器的总内存上限。这里有个容易踩的坑这个值要和 Yarn 队列里允许申请的最大容器内存对比。如果配了 4GB但队列限制容器只能用 2GBYarn 会直接拒绝容器申请作业反复重启。后面我会专门说这个排查方法。taskmanager.numberOfTaskSlots 控制单个 TaskManager 能跑几个任务并发。这个值不是越大越好它主要受 CPU 和内存限制。假设 TaskManager 总内存 4GB、slot 配 4那每个 slot 大概分到 1GB如果你的算子状态很大这个内存根本不够用作业跑起来就会频繁触发垃圾回收。一个合理的经验是slot 数 ≤ 节点可用 CPU 核数内存按每个 slot 至少 1GB 估算。再说两个和 Yarn 绑定的参数。yarn.application-attempts 控制作业失败后最多重启几次默认 -1 表示由 Yarn 决定我通常显式设成 4-5防止无限制重启打爆队列。yarn.application-master.vcores 和 yarn.application-master.memory 决定 AMApplicationMaster的资源配置前者对应 CPU 虚拟核数后者对应容器内存。AM 内存太小会导致提交作业时 AM 频繁重启这个值建议至少 1GB。3.3 提交命令与新旧版本差异配置就绪后就可以提交作业了。先明确一点不同 Flink 版本的提交命令差异不小。老版本统一用 flink run -m yarn-cluster -yjm 1024 -ytm 4096其中 -m 指定运行模式-yjm 指定 JobManager 内存-ytm 指定 TaskManager 内存。新版本1.11 之后把 Per-Job 模式改成 flink run -t yarn-per-jobSession 模式则变成 flink run -t yarn-session。这里有个特别磨人的细节新版本通过 -t 指定目标后-yjm、-ytm 这些简写参数大概率不生效因为新框架推荐把所有配置写进 flink-conf.yaml 或通过 -D 参数覆盖。我实际测试过新版下老老实实编辑 flink-conf.yaml 才是最稳的。提交命令大致长这样以较常见的新版本为例# 进入 Flink 安装目录执行 ./bin/flink run \ -t yarn-per-job \ -D jobmanager.memory.process.size2048m \ -D taskmanager.memory.process.size4096m \ -D taskmanager.numberOfTaskSlots4 \ -c com.example.MyJob \ /path/to/my-job.jar \ --input /data/input.txt --output /data/output还有一点经验生产提交建议先做一次预检用 yarn application -list 看一眼当前队列有没有空闲资源再用 yarn top 或 Yarn Web UI 确认资源水位。资源紧张时提交会一直排队初看以为是卡住了实际在等资源。4. 一个作业从提交到跑完的完整生命周期把部署流程跑通之后强烈建议花点时间把 Flink On Yarn 作业的完整生命周期搞清楚。很多排障问题如果对底层流程有了概念基本一眼就能看出卡在哪个环节。4.1 提交阶段客户端到 ResourceManager作业提交的第一步发生在客户端。Flink 客户端会解析用户代码、生成 JobGraph然后通过 Hadoop 的 YarnClient API 向 ResourceManager 发起应用提交请求。这个请求里面包含了 Flink 的完整打包信息比如运行的 JAR 包、Flink 依赖的分布式缓存文件等。ResourceManager 收到请求后会在某个 NodeManager 上分配一个容器并在容器内启动 ApplicationMaster。注意这个 AM 就是 Flink 集群的“指挥中心”它负责向 RM 申请后续所有资源同时也作为 Flink 集群和 Yarn 之间沟通的桥梁。这一步常见的坑是客户端机器如果缺少某些 Hadoop 配置或者 HADOOP_CONF_DIR 指向了错误目录AM 容器根本拉不起来Yarn 日志里全是 ConnectionException。别急着查 Flink先看 Yarn 侧的日志判断 AM 到底启动没有。4.2 启动阶段Dispatcher 和 JobManager 的诞生AM 启动后会在内部拉起 Flink 的 Dispatcher再由 Dispatcher 创建 JobManager 运行环境。从这里开始Flink 集群就算正式在 Yarn 容器里“活”过来了。JobManager 启动后第一件事就是解析 JobGraph、计算逻辑计划然后向 ResourceManager 申请 TaskManager 容器。Yarn 这边收到 AM 的资源请求后会根据队列配置、资源剩余情况在合适的 NodeManager 上启动 TaskManager 容器。TaskManager 容器启动后会向 JobManager 发送心跳并注册自己的可用 slot。注册完成后JobManager 开始把具体的任务算子分发到各个 slot 上执行。一个简单的单并行度作业启动流程可能只要几十秒并行度大的作业TaskManager 容器是一个个起来的速度慢一些也正常。4.3 运行阶段任务调度、心跳与失败重启作业真正跑起来后稳定性就看 TaskManager 和 JobManager 这两个角色配合得怎么样了。TaskManager 会周期性地向 JobManager 发心跳JobManager 通过心跳把出口的 task status 反馈回去。如果某个 TaskManager 超过设定时间没有心跳JobManager 会判定它死了并在其他可用 slot 上重新调度失败的任务。Flink 自身的检查点机制在这个阶段起着决定性作用。如果启用了 checkpoint作业失败后可以基于最近一次成功快照恢复状态恢复的流程本质上是重新启动任务并加载远端状态。所以部署层面要做的事就是保证 Yarn 容器挂了之后AM 还能确保 JobManager 和 TaskManager 尽快重新被拉起这个重试次数就由 yarn.application-attempts 控制。实际生产里我见过很多作业在”运行-失败-重启-再失败“里死循环。表面现象是集群里多了一个永远重启的作业实际原因往往出在资源配置不合理或者代码里有未捕获异常。遇到这种场景第一件事是看 AM 的日志一般重启原因都写在里面。5. 常见问题与故障排查实录部署 Flink On Yarn 的过程本质上就是一个不断和 Yarn 博弈的过程。下面整理几个我实际踩过的坑以及对应的排查路径希望能帮你少走弯路。5.1 问题速查表现象可能原因排查手段提交作业后一直无反应队列资源不足、等待资源分配yarn application -list 查看状态Yarn Web UI 看资源水位作业提交后秒失败日志里 AM 反复重启AM 内存过小、配置错误查看 AM 容器日志检查 yarn.application-master.memoryTaskManager 一直起不来队列容器内存限制低于 taskmanager 配置查看容量调度器容器最大值降低 taskmanager.memory作业正常运行几分钟后突然重启TaskManager 被 Yarn 杀掉超过心跳超时查看 NodeManager 日志常见为容器内存超限Flink 无法连接 HDFS报 No FileSystem for scheme缺少 hadoop 依赖或配置文件确认 HADOOP_CONF_DIR 指向正确添加 hadoop 依赖5.2 踩坑实录与独家心得第一个坑是容器内存超限被 Yarn 强杀。现象是作业跑得好好的突然所有 TaskManager 同时消失Flink UI 显示连接拒绝。排查 NodeManager 日志会看到 Process killed by YARN这就是容器内存超了 Yarn 的限制。解决办法不是盲目调大 Yarn 的容器上限而是先分析 Flink 自身的内存组成。用 Flink Web UI 的内存面板检查堆内、堆外、直接内存、JVM 元空间的使用量再据此微调 taskmanager.memory.jvm-overhead 等参数。我见过不少团队把问题归咎于代码 bug结果一查是 Netty 直接内存配得太小。第二个坑是 Hadoop 客户端版本和集群版本不一致。早期我用 Flink 官方的 hadoop 依赖包结果集群是某商业发行版两者默认配置路径不一致客户端拿着错误的 core-site.xml 去访问 HDFS权限校验直接失败。后来学到一招把集群上现成的配置文件拷贝到提交机的 HADOOP_CONF_DIR 下再配好相关的认证文件问题立刻解决。这事看起来简单但实际排错花了我大半天时间。第三个坑是检查点存储路径冲突。多个 Per-Job 作业共用同一个 HDFS 检查点目录结果清理旧作业时把另一个作业的状态也删了恢复数据时一脸茫然。现在我的习惯是每个作业单独建检查点目录命名规则直接带上作业名和时间戳再在配置里通过提交参数动态设置简单粗暴又安全。还有一个很隐蔽的问题值得单独说slot 数与并行度的不匹配。假设 TaskManager 配了 4 个 slot作业并行度设成 6Flink 会先等 4 个 slot 用满剩下的并行度要等新的 TaskManager 容器启动后才有位置。如果 Yarn 资源本来就紧张这一步就会一直等。所以提交大并行度作业前一定要算清楚容器数量和 slot 的乘积是否覆盖并行度别等作业卡住了才回头查。6. 资源规划经验与极简军规最后聊点更实际的计算逻辑。部署只是第一步真正让人头疼的是资源怎么给给了又怎么调这些都是我摔过跤之后的领悟。6.1 并行度、slot、内存的联动计算并行度决定任务并发数slot 决定单个 TaskManager 能承载多少并发。它们之间的关系是死的总 slot 数 TaskManager 容器数量 × 每个容器的 slot 数并且总 slot 数必须大于等于作业并行度否则作业就会排队等资源。以我常用的六十四核高配机器为例如果单个容器给 8GB 内存每容器 4 个 slot那一个作业想跑并行度 16就需要 4 个 TaskManager 容器同时存在。内存开销就等于容器数量乘 8GB还要额外加上 JobManager 的 2GB再考虑容器数量 * 某系数留出 AM 和系统开销粗算一下就差不多知道一个作业大概占用多少资源了。计算时我习惯用“倒推法”先定并行度再算 slot 和容器数最后定内存。假如业务要求最坏情况下的吞吐量而单算子吞吐量测试下来约每并行度每秒处理一万条那我需要 10 个并行度便配 2 个容器、每容器 5 个 slot再参考状态大小估算每个 slot 需要的内存量最后确定容器内存。6.2 那些看似无关却要命的参数细节内存分配上有一条线特别容易踩堆内与堆外的平衡。Flink 的内存模型中taskmanager.memory.process.size 是总内存除此之外有不少会额外占内存的模块比如 JVM 元空间、网络缓冲、RocksDB 的直接内存。所以别把 process.size 直接等同于堆大小否则用着用着就会发现内存被隐藏项吃光了然后频繁 GC。另一个我常提的参数是 taskmanager.memory.managed.fraction。这个值控制托管内存比例主要用于 RocksDB 和排序缓冲。如果作业状态很大但算术操作不多这个比例可以适当调小把内存让给堆内能明显改善性能。我调过几次效果立竿见影。最后建议每个线上作业都显式安装监控告警。不只是看 Yarn 层的指标还要看 Flink 自己暴露的 Metrics比如 Checkpoint 时长、作业背压程度、TaskManager 的内存使用率。通过对比这些指标和资源规划后续调优才有依据否则全靠猜。6.3 给新团队的部署建议如果你正准备把 Flink 从开发环境搬到生产环境我的建议是别一上来就追求全功能部署先把一条主线打通再说。第一步用 Per-Job 模式提交一个最简单的 HelloWorld 作业确认 Yarn 能正常拉起集群。第二步接入实际业务数据把检查点和 restart strategy 配置好跑一整天看稳定性。第三步才开始考虑 Session 模式、多队列、权限隔离这些进阶玩法。这个顺序至少能帮你避开 80% 的基础坑。我自己初学的时候就是因为一上来就搞 Session 共享集群结果各种问题交错出现半天定位不到根因。后来老老实实走这条主线反而学得又快又扎实。最后再分享一个我个人的体会Flink On Yarn 的部署本质上不是“装个软件”而是“把计算任务融入现有的资源调度体系”。配置参数、资源规划这些都是细枝末节最关键的是要理解 Yarn 和 Flink 各管什么出了问题知道去哪里找日志、分析哪一环。把这些底层逻辑吃透以后哪怕换到 Kubernetes 上跑 Flink你也能一眼看出两者的相似之处和差异点迁移起来会顺畅很多。