最近在梳理Flink部署相关的资料正好把flinkOnYarn这块单独拿出来写一篇完整的实操记录。这篇文章不讲虚的直接围绕“为什么要用Yarn跑Flink”“两种提交模式怎么选”“配置参数踩的坑”“任务提交失败怎么排查”这几个大家最关心的问题展开。如果你正在准备把Flink接入已有的Hadoop集群或者已经接入了但时不时遇到OOM、JobManager起不来这类问题这篇文章应该能给你省下不少时间。1. 为什么选择Flink On Yarn统一资源调度的现实考量先聊一个经常被忽视的问题Flink有Standalone、Yarn、Kubernetes等多种部署方式为什么大多数公司最终都选了Yarn我在本地环境跑Flink已经够用了为什么非要折腾一个ResourceManager去管理资源最直接的原因是大多数公司的Hadoop集群早就存在了HDFS、Hive、Spark跑得好好的。如果你为了跑Flink单独部署一套Standalone集群就意味着一堆TaskManager要么闲置吃内存要么高峰期不够用还得专门有人去盯它的存活状态。而Yarn本身就是一套成熟的分布式资源调度框架Flink作为Yarn的“租户”按需申请Container、启动任务、释放资源整条链路是自动化的。有一个场景能非常直观地说明问题你白天要跑批处理任务晚上只跑实时计算任务两类任务如果混用Standalone集群你没法让Flink在晚上自动多占一些资源、白天自动让出来。但放在Yarn上你只需要为实时任务设置一个队列Yarn会根据集群整体负载动态分配资源Flink任务结束后Container自动释放完全不用人为干预。1.1 Flink On Yarn解决的核心痛点我把这类部署方式解决的核心问题总结成三点资源利用率Yarn对集群资源做统一调度Flink的任务和Spark的任务可以共用一套Hadoop集群不会出现“Flink集群空转、Spark集群排队”的尴尬局面。按需申请资源Flink任务启动时向Yarn申请指定数量的TaskManager任务结束后自动释放。相比之下Standalone模式则需要手动维护固定数量的TaskManager运维成本高。任务隔离性每个Per-Job模式的Flink作业都有独立的JobManager和TaskManager互不干扰。咱们平时说的“这个任务把集群资源占满了、其他任务跑不动”在Per-Job模式里基本不会发生因为Yarn会通过队列和容器隔离来限制单个任务的最大资源。1.2 部署模式横向对比简单梳理一下三种模式的差异方便没接触过的读者快速形成认知部署模式资源申请方式运维成本适用场景Standalone手动启动固定集群高需独立维护Flink集群本地测试、小规模场景Yarn Session向Yarn申请长驻集群低集群常驻小任务频繁提交、SQL交互式查询Yarn Per-Job每任务独立申请集群低用完即释放大任务、生产环境批量或实时作业Yarn Application主方法在Client端或集群端执行低1.11后推荐生产环境提交Client不再作为瓶颈这里要特别提一句Flink 1.15之后已经逐渐弱化了Per-Job模式推荐使用Application模式。但很多公司的生产环境还停留在老版本而且Per-Job的资源隔离思路仍然值得理解所以这篇文章会以Per-Job为主讲同时补充Application模式的演进逻辑。2. 部署前必须搞定的环境准备与依赖检查很多人在部署flinkOnYarn的时候一头扎进去改配置结果JobManager一直起不来最后发现是Hadoop环境变量没配对。这个环节我专门单独拿出来讲因为根据实际经验80%的部署失败都出在环境准备阶段。2.1 硬件与软件最低要求Flink On Yarn的本质是把Flink发行包提交到Yarn集群上运行所以以下条件必须满足已部署Hadoop集群Yarn和HDFS可用Hadoop版本建议2.7及以上JDK版本1.8或1.11Flink 1.11之后可以跑在Java 11上但生产环境保守选JDK8更稳Flink发行包建议版本与Hadoop版本匹配。如果你用的是CDH这类发行版尤其要注意Hadoop版本兼容性检查环境时先在任意一台能提交任务的客户端机器上执行hadoop version yarn node -list如果这两条命令都能正常输出说明客户端到Yarn集群的链路是通的。2.2 最容易踩的坑HADOOP_CLASSPATH没有配置Flink On Yarn启动时需要把Hadoop相关的classpath传给Yarn的AppMaster。如果你只设置了HADOOP_HOME没设置HADOOP_CLASSPATH大概率会碰到下面这种错误java.lang.NoClassDefFoundError: org/apache/hadoop/yarn/api/ApplicationMasterProtocol这个问题的排查思路其实很简单Yarn通过Flink提供的脚本去拿Hadoop依赖拿不到就有这个报错。最稳妥的解决方法是登录到Flink提交客户端执行export HADOOP_CLASSPATH$(hadoop classpath)然后把这段export写进提交用户的环境变量配置文件比如~/.bashrc或~/.bash_profile避免每次开会话都手动设置。2.3 确认Flink二进制包里的依赖情况Flink官方发布的二进制包分为flink-1.x.x-bin-hadoop28-scala_2.12.tgz和flink-1.x.x-bin-scala_2.12.tgz两种。前者自带了Hadoop 2.8的客户端依赖后者不带任何Hadoop相关jar包需要通过HADOOP_CLASSPATH补充。我个人的习惯是使用不带Hadoop依赖的包通过HADOOP_CLASSPATH动态引入Hadoop依赖。这样做的原因是避免Flink自带的Hadoop jar和集群里的实际Hadoop版本产生冲突。这一点在CDH、HDP这类定制化Hadoop发行版上尤其明显之前遇到过Flink自带的Hadoop 2.8 jar和CDH的RPC协议不兼容导致AppMaster一直无法注册的情况。2.4 搭建最小可用的Yarn测试环境如果你的环境还没有Yarn可以在单机上快速搭一个最小可用的伪分布式Hadoop环境用来测试!-- core-site.xml -- property namefs.defaultFS/name valuehdfs://localhost:9000/value /property property namehadoop.tmp.dir/name value/data/hadoop-tmp/value /property!-- yarn-site.xml -- property nameyarn.nodemanager.aux-services/name valuemapreduce_shuffle/value /property property nameyarn.nodemanager.aux-services.mapreduce_shuffle.class/name valueorg.apache.hadoop.mapred.ShuffleHandler/value /property启动命令顺序是先启动HDFS再启动Yarn$HADOOP_HOME/sbin/start-dfs.sh $HADOOP_HOME/sbin/start-yarn.sh然后通过jps能看到NameNode、DataNode、ResourceManager、NodeManager这几个进程就说明环境OK了。3. 两种核心运行模式的选型逻辑与使用场景Flink On Yarn的模式选择直接决定了你的系统在并发任务多时的表现和运维难度。我见过不少团队一开始为了省事只开了Session模式结果线上任务越来越多互相抢资源最终不得不全部迁移到Per-Job。提前搞清楚模式的差异能帮你少走这一趟弯路。3.1 Session模式的本质常驻集群共享复用Session模式启动后Flink会向Yarn申请一个常驻的Flink集群。这个集群包含一个JobManager和若干TaskManager之后所有提交的任务都提交到这个集群上执行。工作方式Client - Yarn Session (JobManager TaskManager) - 多个Job同时跑在同一个集群里Session模式的优点很明显提交任务速度快因为JobManager已经启动了不需要每次重新申请资源适合任务数量多但单个任务很小的场景比如即席查询、小批量的SQL分析运维上只需要维护一个集群管理简单缺点也致命所有任务共享同一个TaskManager一个任务的OOM可能影响同集群的其他任务任务之间的资源隔离性差大任务可以把整个Session集群资源耗尽TaskManager挂了上面所有任务都受影响恢复成本高启动Session模式的命令yarn-session.sh -n 2 -tm 2048 -s 2 -d参数含义-n申请TaskManager数量-tm每个TaskManager内存MB)-s每个TaskManager的Slot数-d后台运行3.2 Per-Job模式的本质一任务一集群隔离彻底Per-Job模式每次提交任务时都会向Yarn申请一个全新的Flink集群任务结束后整个集群自动释放。这也是很多生产团队最终选择的模式。工作方式Client 提交任务 - 向Yarn申请AppMaster - 启动JobManager - 申请TaskManager - 运行任务 - 结束后释放集群Per-Job模式的优势不同任务拥有完全独立的JobManager和TaskManager互不影响任务级别的资源隔离性好即使某个任务OOM也不会影响其他任务的运行任务结束自动释放资源对集群资源利用率更友好代价是任务启动时多了一步“申请资源和启动AppMaster”的过程所以任务拉起时间比Session模式稍长一般会多个10秒到30秒不等。提交Per-Job任务的命令示例flink run -m yarn-cluster \ -yn 2 \ -ytm 2048 \ -ys 2 \ -yjm 1024 \ -c com.demo.WordCountJob \ /opt/flink-job/my-flink-demo.jar需要注意-m yarn-cluster表示以Per-Job模式提交这个参数在Flink 1.11之后依然兼容老版本但新环境建议直接使用-t yarn-per-job或干脆走Application模式。3.3 Application模式补上了Per-Job的最后一块短板很多人在生产环境中会遇到这样一个问题Per-Job模式下用户代码是在Client端执行的如果用户的main方法里面做了大量本地计算或者连接了外部数据库Client端的内存和网络都可能成为瓶颈。Application模式就是为了解决这个问题而出现的。Application模式同样是一任务一集群但用户代码的执行位置从Client移动到了集群端的JobManager。这意味着客户端提交完任务后可以立即退出不占用本地资源任务的启动开销比Per-Job更小适合长时间运行的流任务不会因为Client断网或重启导致任务提交失败提交Application模式的命令flink run-application -t yarn-application \ -Djobmanager.memory.process.size1024m \ -Dtaskmanager.memory.process.size2048m \ -c com.demo.StreamingJob \ /opt/flink-job/my-streaming-job.jar3.4 如何根据实际场景选型我根据自己的实战经验给出一个相对实用的选型参考业务场景推荐模式理由SQL即席查询、小批量任务、任务数量多Session提交快集群复用生产环境的批处理任务Per-Job资源隔离性好用完即释放生产环境的长时间流任务Application或Per-Job避免Client瓶颈稳定性优先团队多人共享集群按队列划分Session通过Yarn队列控制资源上限大数据量离线任务Per-Job独占资源避免干扰我在实际生产中最常用的是Per-Job和Application两种模式。如果你们团队还处在摸索阶段我建议先跑Session把业务逻辑验证通了再切到Per-Job或Application进行生产部署。4. 实践环节从配置到提交的完整操作步骤这一节直接从零开始把Flink On Yarn的完整启动流程、配置参数含义和提交步骤跑一遍大家照着做就能搭起来。4.1 准备Flink配置文件找到$FLINK_HOME/conf/flink-conf.yaml核心配置如下# JobManager内存配置 jobmanager.memory.process.size: 1024m # TaskManager内存配置 taskmanager.memory.process.size: 2048m # 每个TaskManager的Slot数 taskmanager.numberOfTaskSlots: 2 # 并行度默认值 parallelism.default: 2 # 是否在任务结束后保留Yarn集群Per-Job模式可关 yarn.per-job.keep.containers: false # 自定义Yarn队列 yarn.application.queue: default这几项从字面上不难理解。重点说一下taskmanager.memory.process.size这是TaskManager整个进程的内存上限包含了JVM堆内存和堆外内存。如果给TaskManager分配了4GB的容器Container这个值建议设置为3GB到3.5GB左右留出一些空间给Native层和系统开销否则容器容易超限被杀。4.2 提交Yarn资源队列与容器的关系Yarn中的资源调度以Container为单位Flink的TaskManager在启动时会向Yarn申请指定大小的Container。如果你在flink-conf.yaml中设置了taskmanager.memory.process.size2048m那Yarn分配的Container大小也应该是2GB二者是对应关系。还有一个容易被忽略的参数是yarn.nodemanager.pmem-check-enabled这个参数决定NodeManager是否检查容器物理内存使用量。如果Yarn集群设置了严格的物理内存检查而Flink写入了大量堆外内存比如RocksDB状态后端就可能触发Container被杀。实际解决思路是合理设置TaskManager内存预留20%左右作为堆外预留或者调整taskmanager.memory.off-heap.size手动预留堆外空间如果确实需要大堆外内存可以在Yarn层面适当放宽物理内存检查的阈值4.3 启动Session集群并提交测试任务先启动Session集群cd $FLINK_HOME/bin # 申请2个TaskManager每个2048MB内存每个2个Slot yarn-session.sh -n 2 -tm 2048 -s 2 -d执行后日志会输出类似这样的信息2024-01-15 14:32:45,145 INFO org.apache.flink.yarn.YarnClusterDescriptor - YARN session cluster started 2024-01-15 14:32:45,150 INFO org.apache.flink.yarn.YarnClusterDescriptor - JobManager Web Interface: http://node01:8088/proxy/application_1612345678910_0001/然后往Session集群提交任务flink run -m yarn-cluster \ -p 4 \ -c com.demo.WordCountJob \ /opt/flink-job/my-flink-demo.jar \ --input hdfs:///data/input.txt \ --output hdfs:///data/output这里使用了-p 4指定并行度。在Session模式下如果并行度超过了TaskManager总的Slot数上面配置是2x24个Slot任务会一直等待资源而不是直接报错这一点需要注意。并行度设置过大的情况下提交任务看似成功但Task一直处于SCHEDULED状态直到有TaskManager释放出Slot。4.4 提交Per-Job模式任务Per-Job模式不需要预先启动Session集群直接通过flink run -m yarn-cluster提交即可flink run -m yarn-cluster \ -yn 2 \ -ytm 2048 \ -yjm 1024 \ -ys 2 \ -p 4 \ -c com.demo.WordCountJob \ /opt/flink-job/my-flink-demo.jar \ --input hdfs:///data/input.txt \ --output hdfs:///data/output这几个参数逐一解释下-yn申请TaskManager的个数这里申请2个-ytm每个TaskManager的内存单位MB-yjmJobManager的内存单位MB-ys每个TaskManager的Slot数-p任务并行度任务提交后可以通过Yarn的Web UI观察Application的启动状态。你会先看到状态为ACCEPTED然后JobManager启动完成后变为RUNNING。任务结束后Per-Job模式的Application会变为FINISHED或KILLED容器自动释放。4.5 日志查看小技巧在Yarn上跑Flink最烦的就是不知道日志在哪。推荐的做法是先通过Yarn的Web UI找到Application ID然后命令行查看日志yarn logs -applicationId application_1612345678910_0001这个命令能拿到完整的容器日志包括JobManager和TaskManager的启动日志、用户日志。如果是排查ClassNotFound或OOM这类问题这是最快定位的方式。另外也可以通过Flink的Web UI直接看TaskManager的日志但前提是AppMaster还活着。5. 生产环境中的常见坑与完整排查链路这部分可能是全文最有实战价值的一块因为这些坑基本都是在生产环境折腾了几个小时才总结出来的。5.1 问题一TaskManager启动后反复退出容器被回收现象描述任务提交成功JobManager起来了但TaskManager一直起不来。Yarn的界面上能看到TaskManager的Container状态反复在RUNNING和KILLED之间切换。排查链路第一步查看NodeManager日志或通过yarn logs获取TaskManager容器日志yarn logs -applicationId application_xxx | grep -i error第二步重点查看日志中是否有Container is running beyond virtual memory limits或Container killed by the node manager等字样。第三步如果确认是物理内存超限问题说明TaskManager进程实际内存超出Container限制调整taskmanager.memory.process.size或者在flink-conf.yaml中手动增加堆外内存缓冲。我遇到过最典型的情况是给TaskManager配了2048m内存但状态后端使用的是RocksDBRocksDB本身会使用大量堆外内存最终实际占用超过2GBContainer被NodeManager杀掉。解决方案是给taskmanager.memory.off-heap.size预留至少512MB到1GB的空间。5.2 问题二Per-Job模式提交任务时一直卡在“Restarting”现象描述任务提交上去日志显示JobManager在反复重启查看Yarn界面发现AppMaster一直处于ACCEPTED或SUBMITTED状态。排查链路这种问题大多是Yarn队列资源不足导致的。Per-Job模式每次提交任务都会尝试在指定队列中申请一个AppMaster容器。如果队列的资源上限已经被其他任务占用新的任务就会一直在队列中等待。验证方式登录Yarn ResourceManager的页面查看目标队列的“Used Resources”和“Max Resources”。如果已用资源达到了上限问题基本就定位了。解决方案有几种将任务提交到资源空闲的队列降低-yjm和-ytm的配置值让任务能在剩余资源中跑起来如果公司有多个计算团队建议为每个团队配置独立的Yarn队列避免互相占用资源5.3 问题三类冲突导致作业初始化失败现象描述任务提交成功但在JobManager的日志中看到NoClassDefFoundError或ClassNotFoundException大多涉及到Hadoop或第三方依赖。排查链路第一种情况是缺少Hadoop依赖。解决办法是配置HADOOP_CLASSPATH并把相关依赖加入环境变量。第二种情况是依赖冲突常见于Flink自带的Hadoop jar与集群的Hadoop版本不一致。处理思路是尽量使用不带Hadoop依赖的Flink二进制包或者把Flink里与Hadoop冲突的jar包替换掉。第三种情况是用户代码里引入了与Flink内部类冲突的库比如引入了老版本的commons-collections或guava。排查方法是在本地用mvn dependency:tree分析依赖树找到冲突的jar然后在pom里做exclusion排除。5.4 问题四中文乱码与时区问题在Yarn容器中跑Flink还会遇到一个不起眼但很烦的问题任务日志或输出结果出现中文乱码或者时间差8小时。这通常是因为Yarn容器运行时没有继承客户端的LANG和TZ环境变量。解决办法是在启动脚本或Flink配置中显式设置export LANGzh_CN.UTF-8 export TZAsia/Shanghai如果提交任务时用的是Flink SQL建议在代码里显式设置时区SET table.local-time-zone Asia/Shanghai;5.5 历史服务器不要忽略的运维必备组件Yarn模式还有一个运维层面的细节——Flink HistoryServer。Per-Job模式下任务结束后整个Yarn集群会被回收任务运行时的指标、日志、Event信息随之消失。如果不配置HistoryServer你事后就看不到这个任务当时的运行情况了。配置方法# conf/flink-conf.yaml historyserver.web.address: 0.0.0.0 historyserver.web.port: 8082 historyserver.archive.fs.dir: hdfs:///completed-jobs/然后启动$FLINK_HOME/bin/historyserver.sh start配置完成后任务结束后JobManager会把Event文件写入HDFS的/completed-jobs/目录HistoryServer会定期加载这些文件供你查询。这个功能尤其是排查“昨天跑失败的任务”时必不可少。6. 内存参数调优的实用模板既然讲到了flinkOnYarn内存参数的配置是很多新手最容易看懵的地方。我把自己常用的一套生产模板直接贴出来并说明每个参数的定位方便大家作为基线参考。6.1 各内存模块的角色定位Flink进程内存大致分为三块JVM堆内存Heap主要用于用户代码、状态存储、网络缓冲区中的部分数据托管内存Managed Memory主要用于RocksDB状态后端、排序缓冲区等Flink内部数据堆外内存Direct Memory Native用于网络缓冲、框架内部开销这三个模块的关系可以简单类比为堆内存是你租的房子里的客厅托管内存是卧室堆外内存是阳台。客厅小了住得挤卧室小了放不下床阳台完全没有衣服就没地方晾。三块都需要有合适的大小。6.2 典型配置参考对于单个TaskManager内存类型配置项参考值总进程内存taskmanager.memory.process.size3072mJVM堆taskmanager.memory.fraction0.7自动计算托管内存taskmanager.memory.managed.fraction0.2默认堆外内存taskmanager.memory.off-heap.size512m如果是使用RocksDB状态后端的流任务建议把taskmanager.memory.managed.fraction调高到0.3左右给RocksDB留出足够的空间。对于JobManager内存类型配置项参考值总进程内存jobmanager.memory.process.size1024mJVM堆jobmanager.memory.heap.size768mJobManager主要负责调度和Checkpoint协调一般不需要太大的内存。6.3 关于容器大小与内存的关系在Yarn上容器分配的内存大小必须大于等于Flink进程的内存需求。如果taskmanager.memory.process.size3072m那么Yarn申请的Container大小最好设置为3GB或3.5GB。我见过有人配的时候只设置了taskmanager.memory.process.size但忽略了yarn.containers.vcores导致CPU资源分配不足任务运行速度很慢。建议在提交时配合-yn和-ys时也要注意每个容器的vCores配置默认是1如果任务计算量大可以适当调高。7. 从Per-Job迁移到Application模式一次真实的上线经历最后用一段个人经历来补充说明为什么我后来在生产环境更推荐Application模式。早前处理一个实时ETL任务上游数据量在高峰期会翻好几倍任务需要较长的启动时间。一开始用的是Per-Job模式某天因为客户端所在的机器由于网络策略调整导致Client到Yarn集群的连接中断Per-Job模式的任务提交过程因为Client端main方法还在执行结果整个作业提交失败。如果是流任务重启这一断就丢了半小时的数据。后来把该任务迁移到Application模式后Client端只负责上传jar包和触发提交随后的流程全部由集群端完成本地断网不再影响任务运行。对于需要长时间稳定运行的流任务这种模式在运维上能省很多精力。迁移的大致步骤确认Flink版本在1.11以上将提交命令从flink run -m yarn-cluster改为flink run-application -t yarn-application将-yn等参数改为通过-D传参例如-Dtaskmanager.numberOfTaskSlots2测试本地的提交脚本确认参数是否生效flink run-application -t yarn-application \ -Djobmanager.memory.process.size1024m \ -Dtaskmanager.memory.process.size3072m \ -Dtaskmanager.numberOfTaskSlots2 \ -Dparallelism.default6 \ -Dyarn.application.queuerealtime \ -c com.demo.RealtimeETLJob \ /opt/flink-job/realtime-etl.jarApplication模式下的日志查看方式和Per-Job模式一致仍然使用yarn logs -applicationId来获取这一点没有变化。最后再多说一句Flink On Yarn的版本选择建议。如果你现在才开始搭建尽量选择1.14以上的版本因为Per-Job和Session模式的代码路径逐渐收敛Application模式愈发成熟。如果公司已有老版本作业建议按作业维度逐步迁移不要一次全部切到Application毕竟每种模式的资源申请流程和故障恢复细节存在差异。我自己的习惯是按“批处理用Per-Job、长时间流任务用Application”这个思路分流运维起来会比较清爽。