★ FEATURED ARTICLE 我帮人排查过不少 Flink on YARN 集群的问题最后发现十有八九不是 Flink 本身多复杂而是部署环节的某个细节没对齐版本选错了一个包、HADOOP_CLASSPATH没配、TaskManager 要的内存超过了 YARN 的资源上限。这篇不想讲太多理论直接按我自己的部署流程来用什么包、改哪几个配置、跑什么命令、验证哪些东西、出了问题按什么顺序查。目标是让你照着操作能把集群撑起来并且稳定提交作业。假设你已经有一套 Hadoop/YARN 集群在手JDK 也装好了下面都是实操。1. 部署模式先选对后面能少走一半弯路Flink 跑在 YARN 上有 Session、Per-Job、Application 三种模式很多人一开始没想清楚就装装完发现我这个模式是不是选错了来回折腾。我先用一句话把三种模式的核心区别说清楚。1.1 三种模式一张表看懂模式集群生命周期适用场景资源隔离运维复杂度Session常驻一次启动多次提交短任务、交互式查询、开发调试共享 JobManager 和 TaskManager低Per-Job经典模式每个作业一个集群作业结束集群销毁长时间运行的大作业、资源敏感作业完全隔离中Application推荐每个应用一个集群main()在集群内执行生产环境标准作业完全隔离中Session 模式就是你启动一个yarn-session得到一个常驻的 Flink 集群之后往这个集群里提交多个作业。适合小作业、作业数量多、对延迟敏感的场景。缺点也明显多个作业共享资源一个作业把资源吃满了其他作业就排队。Per-Job 模式是 Flink 1.15 之前的主流提交作业时单独起一个 YARN Application作业结束集群就退房。好处是每个作业完全隔离坏处是main()逻辑在客户端执行客户端断了或者压力大了会影响整个提交过程。Application 模式是我现在生产环境里最常用的。它把 Per-Job 的客户端main()搬到了集群内部执行客户端只需要提交一个命令然后就可以不管了。这样既保留了 Per-Job 的资源隔离又解决了客户端单点的问题。1.2 为什么我建议第一次部署先跑 Session不管是生产还是测试环境我第一次接触一套新集群永远先把 Session 模式跑通。原因很简单Session 模式启动最简单验证链路最短。一个 Session 集群起来之后JobManager 和 TaskManager 都在 YARN 上注册好了你再提交任何作业只需要看作业本身的日志和状态不用把集群部署问题和作业运行问题混在一起排查。后面你完全可以在同一套 Flink 安装目录下跑 Application 模式命令就一行不需要重新部署。这也是我把 Session 的完整链路先讲清楚的原因——先有一套能用的集群再谈其他模式。至少我接触过的团队里还没有哪次是Session 能跑通但 Application 模式装不起来的两者底层依赖是一样的。2. 版本匹配这关过不去安装包再新也白搭Flink 安装包本身不复杂复杂的是它和 Hadoop、YARN 的版本关系。很多人在这一步踩坑下载了一个最新版 Flink结果和集群的 Hadoop RPC 协议对不上作业反复失败。2.1 Flink 与 Hadoop 版本的对应思路Apache Flink 的官方下载包长时间以来分成两类带 Hadoop 依赖的预编译包和不带 Hadoop 依赖的包。带 Hadoop 的包是把某一个大版本 Hadoop 的flink-shaded-hadoop打进去了比如 2.10 或者 3.x。如果你的集群 Hadoop 是另一个版本很可能在运行时出现NoSuchMethodError、NoClassDefFoundError这类莫名其妙的错误。我的建议很明确优先选不带 Hadoop 的 Flink 包通过HADOOP_CLASSPATH指向集群自身的 Hadoop 客户端依赖。这样 Flink 用的就是和集群一模一样的 Hadoop 库从根上避免版本冲突。实际操作时你需要做两件事下载时认准包名里带without hadoop或类似标识的安装包。在每台节点上配置HADOOP_CLASSPATH让 Flink 启动时能找到集群的 Hadoop 依赖。2.2 HADOOP_CLASSPATH 的标准配置方法如果你用的是 CDH 发行版或自己编译的 HadoopHADOOP_CLASSPATH不是一个默认存在的环境变量需要你手动拼出来。最省事的办法是借助 Hadoop 自带的hadoop classpath命令# 先确认 hadoop 命令在 PATH 里 hadoop version # 把 classpath 追加到 /etc/profile 或 ~/.bashrc export HADOOP_CLASSPATH$(hadoop classpath)如果你手头有hadoop-client、hadoop-hdfs、hadoop-yarn这些 jar 包的目录也可以手动拼但hadoop classpath命令最稳定它会把你集群里实际生效的所有依赖目录都列出来。配置完之后一定要验证我吃过亏echo $HADOOP_CLASSPATH如果输出为空串说明环境变量没生效。Flink 提交时会报类似 No ClassFound org.apache.hadoop.conf.Configuration 的错误排查起来特别绕因为你明明确认过 Hadoop 本身能用。2.3 Java 版本和 Scala 版本也要对齐Flink 安装包名里一般会有_2.12或_2.13这样的后缀这是 Flink 编译用的 Scala 版本。你的作业如果用 Scala API 开发Scala 版本必须和 Flink 保持一致如果用 Java API影响不大但还是建议选一个社区主流的组合。另外Flink 1.17 之后的版本要求 JDK 8 或 JDK 11 都可以很多新特性在 JDK 11 上测试更充分。如果你的集群还跑着老旧的 JDK 8 小版本建议先升级到较新的 8u 版本否则某些 TLS、集合类的问题会间歇性出现非常难查。3. 资源规划怎么算YARN 内存上下限是硬门槛很多人在flink-conf.yaml里配内存时很随意结果启动 Session 集群后TaskManager 反复注册失败。根本原因是你给 Flink TaskManager 的内存大小超出了 YARN 单个 Container 允许的范围。3.1 YARN 资源调度的三个关键参数YARN 本身就是 Flink 的房东它说了算。真正约束你 Flink 能否起来的是yarn-site.xml里的这几个参数yarn.nodemanager.resource.memory-mb每个 NodeManager 节点能分配给所有 Container 的总内存。yarn.scheduler.minimum-allocation-mb单个 Container 的最小内存默认一般是 1024。yarn.scheduler.maximum-allocation-mb单个 Container 的最大内存默认一般是 8192。Flink 的 JobManager 和 TaskManager 在 YARN 上各自对应一个 Container。它们的进程总内存jobmanager.memory.process.size和taskmanager.memory.process.size必须落在 min 和 max 之间不然 Container 请求直接失败YARN 会反复重试。你可能觉得 8G 上限很大了但如果你给 TaskManager 配了 10G作业就会一直卡在 ACCEPTED 状态。这是我见过的最常见的集群起不来原因之一。3.2 一套 4 节点的内存分配参考假设每台机器物理内存 64GHDFS 的 DataNode 和 YARN 的 NodeManager 都在这台机器上那么我一般这样分配组件分配内存说明操作系统及系统服务8G留足 page cache 和系统进程HDFS DataNode4G根据实际负载调整YARN NodeManager 可分配48Gyarn.nodemanager.resource.memory-mb49152YARN container 上限24Gyarn.scheduler.maximum-allocation-mb24576注意yarn.scheduler.maximum-allocation-mb我一般会设成一个比较大的值比如 24G 或 32G给 Flink 的 TaskManager 留弹性。如果你不确定保守一点 16G 也够用但必须保证大于你希望的 TaskManager 内存。Flink 侧对应配置jobmanager.memory.process.size: 2048m taskmanager.memory.process.size: 8192m taskmanager.numberOfTaskSlots: 4 parallelism.default: 2TaskManager 进程总内存 8G落在 YARN 的 16G 上限内同时单节点可以再跑多个 TaskManager Container灵活性就出来了。3.3 给系统留内存是我多年养成的习惯有一种部署误区是内存全部分给 YARNFlink 想吃多少吃多少。实际结果是操作系统本身内存不足开始疯狂 swapDataNode 心跳超时整个集群变得极不稳定。我给每个节点至少留 15% 的物理内存给系统和 HDFS尤其是页缓存。Flink 处理数据大量依赖磁盘和网络留足页缓存对吞吐影响非常大。这条建议你可以认为保守但集群稳定性和数据安全比那几 G 内存值钱得多。4. Flink 配置文件不用大改但这几项必须对齐Flink 的配置项有几百个但部署一个能用的集群需要改的其实就那十几个。很多人一上来各种参数调一晚上最后发现根因只是内存没对齐。4.1 flink-conf.yaml 里的核心配置修改conf/flink-conf.yaml逐项说明我的配置逻辑# 总内存配置与 YARN 的资源上限对齐 jobmanager.memory.process.size: 2048m taskmanager.memory.process.size: 8192m # 每个 TaskManager 的 slot 数量我一般配成 CPU 核数 taskmanager.numberOfTaskSlots: 4 # 默认并行度保守一点先从 2 开始跑 parallelism.default: 2 # JobManager Web 端口显式指定避免随机端口冲突 rest.port: 8081 # 如果有 ZooKeeper做 HA 的最低配置 high-availability: zookeeper high-availability.zookeeper.quorum: node01:2181,node02:2181,node03:2181 high-availability.storageDir: hdfs:///flink/ha关于jobmanager.memory.process.size和taskmanager.memory.process.size这两项代表的是进程总内存包括 JVM 堆、堆外、网络缓冲、JVM overhead 等所有部分。YARN 分配 Container 内存看的就是这个总大小。不要只看 JVM 堆大小否则你堆配 4G、总进程配 4G最后 overhead 超了照样起不来。taskmanager.numberOfTaskSlots设置为 CPU 核数是行业通用经验但不绝对。如果每个 slot 处理的作业都比较重适当减少到核数的一半也没问题。4.2 masters 和 workers 文件怎么填在conf/masters文件里写 JobManager 所在的主机名node01如果做 HA可以写多个每行一个。Session 模式和 Application 模式对 masters 的处理不太一样Session 模式会启动一个真实 JobManager 进程所以集群里这台机器最好能固定下来。在conf/workers文件里写 TaskManager 所在的主机名一行一个node02 node03 node04这里要注意workers 文件里每行一个节点不要写两个空格不要带多余换行。我用过不少发行版偶尔会有因为文件末尾多了个不可见字符导致节点没被识别的情况。改完之后可以用cat -A看一眼有没有^M这类特殊字符。4.3 环境变量脚本bin 目录下容易被忽略的配置conf/flink-env.sh这个文件也很关键。如果它存在你需要保证JAVA_HOME/usr/local/jdk HADOOP_CLASSPATH$(hadoop classpath)如果不写 JAVA_HOMEFlink 启动时会用 PATH 里的 java可能不是你期望的版本。多个 JDK 并存的时候这个问题尤其隐蔽你明明java -version是 11但 Flink 起了一个 JDK 8 的进程然后一堆兼容性报错。5. Session 集群完整启动流程命令和验证一次性给齐配置改完之后真正的部署流程其实很短但每个步骤都要稳。下面是我在一套三节点集群上的完整操作顺序。5.1 安装包分发与解压先把 Flink 安装包上传到第一台节点比如放到/opt目录cd /opt tar -zxvf flink-1.17.2-bin-scala_2.12.tgz mv flink-1.17.2 flink然后用 scp 同步到其他节点scp -r /opt/flink node02:/opt/ scp -r /opt/flink node03:/opt/这里有个小建议软链到一个统一路径比如/opt/flink目录名不带版本号。这样将来升级时只需要切换软链不用改一堆脚本里的路径引用。所有节点的/etc/profile统一追加export FLINK_HOME/opt/flink export PATH$PATH:$FLINK_HOME/bin5.2 SSH 免密登录检查Flink on YARN 部署不强制要求 SSH 免密但如果你要用flink命令在每个节点之间协调还是建议配好ssh node02 hostname ssh node03 hostname如果这里不通后续start-cluster.sh这种脚本会失败得很隐晦。不过话说回来Flink 真正跑起来后Container 的启停都是 YARN 在做SSH 主要是给部署和运维省事不是关键路径。5.3 启动 Session 集群在任意节点执行yarn-session.sh -d -nm flink-session-demo参数说明-d后台运行不会挂在当前终端。-nm flink-session-demo给 YARN Application 起一个便于识别的名字。启动成功之后控制台会输出类似这样的信息Flink Session Cluster with ID xxx has been started.这时候立刻用 YARN 命令验证yarn application -list你应该能在输出里看到一个状态为 RUNNING 的 application名字是flink-session-demo。如果状态反复在 ACCEPTED 和 RUNNING 之间切换大概率是资源不够或者内存超了回到第 3 节检查 YARN 参数。5.4 Web UI 验证YARN 命令能查到 RUNNING 还不够我习惯继续查一下 Flink 自带的 Web UI。在yarn application -list输出里找到这个应用对应的 URL形如http://node01:8081。浏览器打开能看到 JobManager 的 Overview 页面左侧任务列表为空但 Task Managers 菜单里能看到已注册的 TaskManager 列表。如果 Web UI 打不开优先查两件事YARN 上那个 application 所在节点的本地防火墙以及 JobManager 所在机器的 8081 端口是否被占用。很多团队把节点安全组限制得很严格这个端口忘了放行调试起来会特别莫名其妙。6. 第一个作业跑起来的完整验证链路Session 集群活着不等于真的能用必须实际提交一个作业验证全链路。Flink 自带 examples 包里有现成的 WordCount我就拿它演示。6.1 提交自带 WordCount 作业进入 Flink 解压目录执行flink run -d \ -m yarn-session \ -p 2 \ examples/streaming/WordCount.jar-m yarn-session是告诉客户端去连已经启动的 Session 集群-p 2显式指定并行度。不传-p的话会使用parallelism.default的配置。提交后再次打开 Web UI刷新 Job 列表能看到一个 RUNNING 的作业。点进去能看到数据流图和每个算子的并行子任务分布。正常情况下2 个并行度会分散到不同 TaskManager 的 slot 上。6.2 确认日志和反压状态Web UI 右上角可以看 TaskManager 日志。点进作业详情先看几点有没有反压。如果 Source 和 FlatMap 算子出现红色说明下游处理不过来可能是并行度不够。每个 TaskManager 的 heap 使用是否平稳。如果一个 TaskManager 内存峰值很高考虑加大taskmanager.memory.process.size。这些信息不用等到作业跑完再判断。在线看日志是准的因为 YARN 的 log aggregation 默认是作业结束后才集中运行中看本地日志最方便。6.3 停作业和停集群作业跑完通过flink stop jobId可以优雅停止。想停掉整个 Session 集群执行yarn application -kill applicationId或者直接杀掉yarn-session.sh对应的进程。生产环境不建议用 kill 方式停 Session因为 JobManager 需要时间做状态清理粗暴杀掉可能留下异常记录。如果你要跑 Application 模式提交方式完全不同flink run-application -t yarn-application \ -Dyarn.application.nameMyApp \ -Dyarn.application.queuedefault \ your-job.jar这个模式下没有常驻集群每个命令都会启动一个新的 YARN Application。我习惯在生产作业里用这种方式因为它天然隔离并且yarn application -list就能看到所有作业的运行状态。7. 部署完最容易翻车的 5 类问题与排查顺序最后这部分是干货中的干货。以下问题我都在真实环境里见过写成症状 排查顺序 解决的形式方便你以后直接对照。7.1 NoClassDefFoundErrorHadoop 相关类找不到症状Session 集群能起来但提交作业后 JobManager 或 TaskManager 日志里报NoClassDefFoundError指向org.apache.hadoop.conf.Configuration之类的类。排查顺序在所有节点执行echo $HADOOP_CLASSPATH确认不是空值。确认/etc/profile里的 export 在flink-env.sh里也生效了。检查 flink-env.sh 里是否有重复的配置覆盖了你的环境变量。我遇到过一种很隐蔽的情况/etc/profile配了HADOOP_CLASSPATH但 Flink 启动时用的 shell 是 non-login shell不会读/etc/profile导致客户端提交时正常、YARN 里的进程找不到依赖。解决方法是在flink-env.sh里显式写一遍 export。7.2 TaskManager 反复启动失败内存超 YARN 配额症状YARN UI 里 application 一直 RUNNING但 TaskManager 数量始终是 0日志里频繁出现 Container launch failed 或 Resource limit exceeded。排查顺序先看 YARN 的 ResourceManager 日志通常写在yarn.log-aggregation-enable配置的目录或容器日志里。比较taskmanager.memory.process.size与yarn.scheduler.maximum-allocation-mb的大小关系。用yarn node -list -all查看单个节点剩余资源。这个坑的破案速度取决于你对第 3 节资源规划上不上心。我通常不改任何业务代码直接调小 TaskManager 内存到 YARN 上限以内再启动一次问题当场消失。7.3 本地能跑、集群上编译不过或缺类症状IDEA 里跑 Flink 作业一切正常丢到集群上就报各种 ClassNotFoundException比如 Kafka 连接器、数据库驱动。原因Flink 官方 jar 包和本地依赖有差异或者你的作业 jar 没有把第三方依赖打进去。解决方案plugin groupIdorg.apache.maven.plugins/groupId artifactIdmaven-shade-plugin/artifactId version3.5.1/version executions execution phasepackage/phase goalsgoalshade/goal/goals /execution /executions /plugin用 shade 插件打成 fat-jar 再提交。还有一种偷懒但可行的方案把第三方 jar 放到 Flink 的lib目录下但要谨慎防止类冲突。7.4 作业一直 PENDING不进入 RUNNING症状作业提交后状态一直 PENDINGWeb UI 里没有任何 TaskManager 收到任务。原因通常有两个集群并行度大于实际可用的 slot 总数。队列资源不足YARN 在等资源。排查顺序数一下taskmanager.numberOfTaskSlots乘以 TaskManager 数量。在 Web UI 查看 slot 是否全部空闲。看 YARN 队列资源使用情况。我在测试环境里最常用的做法是先手动指定-p 1提交如果能跑起来再逐步调大并行度快速定位是资源不够还是代码本身无法并行。7.5 多个 Session 集群端口冲突症状开了多个 Session 集群第二个集群的 Web UI 打不开日志里提示8081 port already in use。原因rest.port固定为 8081而新的 JobManager 也尝试绑定 8081。解决办法在flink-conf.yaml里把rest.port改成 0让端口随机分配需要时从 YARN 的 Tracking UI 里看实际端口。或者显式指定不同的端口比如第二个 Session 用-Drest.port8082。多 Session 并行在测试环境往往是为了隔离不同团队的作业但生产环境不建议这样玩用 Application 模式更干净每个作业独立 Application ID自然不会撞端口。7.6 日志定向运行中要防日志飘走很多人在 YARN 集群上找不到日志是因为 YARN 的 log aggregation 会把日志聚合到 HDFS 上默认只在作业结束后归档。运行中的日志在容器本地目录例如yarn logs -applicationId appId这样看的是聚合后的日志可能滞后。运行中更直接的办法是登录 NodeManager 所在节点看/opt/logs/下的容器日志目录。建议从第一天部署就把 Flink 的日志路径固定到统一目录比如/var/log/flink并在log4j.properties或log4j2.properties里按 JobManager 和 TaskManager 分文件这样排查问题会省力很多。收尾一点小建议我在实际部署中养成的习惯是每次改完flink-conf.yaml先做一次diff保存标清楚改了哪几项、为什么改。因为 Flink 配置文件项目非常多过了三个月你再回来看根本记不清哪项是默认的、哪项是为了解决某个特殊问题才加上去的。把部署过程写成一段简单的初始化脚本放在代码仓库里新环境直接跑一遍比手动敲命令可靠得多。集群部署这件事难的不是命令本身而是把环境变量、版本、内存、权限这些琐碎细节一次性对齐少一个都不行。希望这篇不废话的部署记录能让你少走几趟弯路。
阅读完成 · 觉得有帮助？
