Flink StandAlone模式作业提交全流程实战指南-北京尧图网络

做实时计算这行跟Flink打交道是绕不开的。如果你刚接触Flink或者正在纠结怎么把手里的作业跑起来我建议你从StandAlone模式入手。很多人在学习阶段就直接上YARN或者K8s结果被资源管理、容器调度这些概念绕晕了反而把核心的作业提交逻辑给忽略了。StandAlone模式是Flink自带的独立部署方式不需要依赖任何外部资源调度框架一台机器或者几台机器就能搭出一个完整的集群。你可以用它来学习、测试也可以在小规模生产环境里跑一些轻量级的实时任务。这篇文章会从零开始把StandAlone模式下提交Flink作业的完整流程走一遍。内容包含集群的架构设计、环境准备、配置文件调整、作业打包、命令行提交、Web UI提交、SQL Client提交以及我在实际运维中踩过的坑和排查思路。不论你是刚入门的菜鸟还是已经接触过Flink但没系统梳理过提交流程的同学这篇文章都能给你一个可以直接照做的参考。1. StandAlone模式整体架构与设计思路1.1 为什么先选StandAlone模式我见过不少刚接触Flink的开发者一上来就直奔YARN或者Kubernetes觉得那是生产环境的标准答案。这个想法本身没错但问题在于当你对Flink的作业生命周期、任务调度、检查点机制这些都还没有直观概念的时候再叠加上一套外部资源管理框架排查问题的时候你会分不清到底是Flink的问题还是资源框架的问题。StandAlone模式最大的价值是把Flink本身的功能边界画得清清楚楚。它只用Flink自带的组件来完成资源管理与任务调度不需要额外的依赖。你在一台笔记本上就能启动一个最小集群然后观察一个作业从提交、调度、执行到完成的全过程。对于学习来说这比任何文档都直观。这套模式在小规模生产环境里也有用武之地。比如你有一个数据量不大的实时ETL任务每天处理几百GB数据用两三台机器搭一个StandAlone集群完全够用。它的运维成本很低不需要维护额外的资源调度服务出问题的时候定位路径也比较短。1.2 核心组件与角色划分StandAlone集群由两大角色构成JobManager和TaskManager。JobManager是集群的“大脑”负责接收作业、调度任务、协调检查点、处理故障恢复TaskManager是“工人”负责真正执行算子逻辑、缓存数据、与上下游交互。在Flink 1.11之后的版本里JobManager内部进一步拆分成了Dispatcher、ResourceManager和JobMaster三个逻辑组件。Dispatcher负责接收用户提交的作业并提供REST接口供Web UI和命令行调用ResourceManager负责管理TaskManager的槽位资源作业需要多少槽位它就协调多少JobMaster则具体管理单个作业的整个生命周期。你可以这样理解Dispatcher是前台接待ResourceManager是后勤调度JobMaster是项目负责人TaskManager是一线干活的人。这个类比虽然粗糙但能帮助你快速建立整体印象提交作业到DispatcherDispatcher通知ResourceManager分配槽位JobMaster负责把作业调度到具体的TaskManager上执行过程中任何一个环节出问题日志都会指向对应的组件。1.3 StandAlone与YARN、K8s模式选型对比选哪种部署模式本质上是在回答一个问题谁来管理计算资源StandAlone模式下Flink自己管理资源。集群启动后TaskManager的槽位是固定好的作业提交后只能在这些槽位里调度。优点是部署简单、概念清晰缺点是资源利用率低集群闲的时候浪费忙的时候可能不够。YARN模式把Flink作业放在Hadoop的YARN集群里由YARN动态分配容器作业结束后资源自动释放适合跟Hadoop生态深度绑定的场景。K8s模式则是通过容器编排平台来管理Flink集群和作业弹性最好适合云原生环境但学习和运维成本也是最高的。对比项StandAloneYARNKubernetes部署复杂度低中高资源动态分配不支持支持支持学习门槛低中高适合场景学习、测试、小规模生产Hadoop生态生产环境云原生大规模生产故障恢复依赖Flink自身YARN重新拉起K8s重新调度我当时第一次在生产环境用Flink就是从StandAlone开始的。三台机器每台跑一个TaskManager跑了一个实时指标计算任务一年下来几乎没有因为集群本身出过问题。它不像大家说的那么“玩具”关键是你要把资源规划好并且接受它没有动态伸缩这个事实。2. 环境准备与集群启动2.1 环境要求与JDK版本选择StandAlone模式对硬件的要求不高最低配置一台2核4G的机器就能跑起来。但如果你要跑真实的计算任务我建议最少三台机器一台跑JobManager两台跑TaskManager避免单点故障导致作业直接挂掉。机器操作系统选Linux最省心CentOS 7或Ubuntu 20.04都行Windows上也能跑但遇到网络和权限问题的概率会高不少。JDK版本是第一个容易踩坑的地方。Flink 1.13及之前的版本对JDK 8支持得最好Flink 1.14开始支持JDK 11Flink 1.18开始支持JDK 17。我的建议是如果你用的是Flink 1.17及以下版本老老实实用JDK 8如果你用Flink 1.18以上版本JDK 8和JDK 11都可以但一定要保持所有节点JDK版本一致。公司里有同事曾经因为JobManager用的JDK 8、TaskManager用的JDK 11导致作业提交后ClassNotFoundException排查了一下午。确保所有节点之间网络互通hostname能互相解析。最简单的方式是配置/etc/hosts把JobManager和TaskManager的主机名和IP映射都写进去。这个步骤看似琐碎但漏掉它后面TaskManager注册不上来的时候你就会很痛苦。2.2 flink-conf.yaml核心配置项解析Flink的配置文件在conf目录下核心是flink-conf.yaml。这个文件决定了集群的资源行为和运行参数。下面这几个配置项是你启动StandAlone集群之前必须搞清楚的。# JobManager进程总内存包含堆内、堆外、JVM自身开销 jobmanager.memory.process.size: 1600m # TaskManager进程总内存 taskmanager.memory.process.size: 2048m # 每个TaskManager可提供的任务槽位数量 taskmanager.numberOfTaskSlots: 4 # 默认并行度 parallelism.default: 1 # 故障恢复策略 jobmanager.execution.failover-strategy: regionjobmanager.memory.process.size和taskmanager.memory.process.size控制的是进程级别的总内存Flink会在内部进一步拆分成堆内、托管内存、网络缓冲等。新手最容易犯的错误是只给TaskManager设置很小的内存比如512M结果作业一跑起来就频繁Full GC性能惨不忍睹。对于普通的流式作业我建议TaskManager至少给2G如果你有窗口聚合、状态存储这类需求4G起步。taskmanager.numberOfTaskSlots决定了一个TaskManager能跑多少个并行子任务。槽位数量不等于物理核数你可以把槽位理解成“执行位”。一个4核8G的机器配置4个槽位是合理的选择。槽位配多了每个任务分到的资源变少反而容易出问题。如果你的作业并行度是2两个TaskManager各4个槽位那么Flink会把作业的两个并行实例尽量分散到不同的TaskManager上。注意parallelism.default只是默认值作业提交时通过-p参数指定的并行度优先级更高。别以为配置文件里写死就万事大吉。2.3 启动集群与Web UI验证配置完成后在JobManager节点上执行启动脚本# 进入Flink安装目录 cd /opt/flink # 启动StandAlone集群 bin/start-cluster.sh如果你之前修改过workers文件在conf目录下每行一个TaskManager主机名脚本会自动在所有Worker节点上启动TaskManager进程。第一次启动时我建议你把每个节点的日志都检查一遍。JobManager的日志在logs/flink-{user}-standalonesession-{id}-{host}.logTaskManager的日志在logs/flink-{user}-taskexecutor-{id}-{host}.log。启动后打开浏览器访问http://jobmanager:8081如果你在本地就是http://localhost:8081你会看到Flink的Web UI。重点看两个地方左侧的Task Managers页面确认所有TaskManager都成功注册并显示你配置的槽位数量Overview页面的Task Slots字段确认总的可用槽位数符合预期。如果TaskManager没有出现在Web UI上最常见的两个原因一是hostname解析不了二是TaskManager的RPC端口默认6122被防火墙拦截。第一个问题去检查/etc/hosts第二个问题检查安全组和firewalld规则。你还可以在JobManager节点上手动执行下面的命令确认Worker到JobManager的网络是否通# 在TaskManager节点上执行JobManagerHost换成实际主机名 telnet JobManagerHost 6123端口通了再去看日志别一上来就怀疑配置。集群起不来时九成是网络剩下才是配置。3. 作业打包与提交方式详解3.1 作业代码的核心思路先写一段最简的流式作业作为演示。这个作业从Socket读取文本按行做词频统计输出到标准控制台。它不涉及外部存储目的是让你把注意力完全放在提交流程上。import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.util.Collector; public class SocketWordCount { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); DataStreamString text env.socketTextStream(localhost, 9999); text.flatMap(new FlatMapFunctionString, Tuple2String, Integer() { Override public void flatMap(String line, CollectorTuple2String, Integer out) { for (String word : line.split(\\s)) { out.collect(Tuple2.of(word, 1)); } } }) .keyBy(value - value.f0) .sum(1) .print(); env.execute(Socket Word Count); } }这里有个细节很多人会忽略env.execute(Socket Word Count)这一行才是真正把作业提交给集群的入口。如果注释掉这行程序会在本地执行算子链但不会触发作业的实际提交。在实际工程中你的代码可能非常复杂但提交的入口永远是这一句。另外socketTextStream这种数据源只适合本地测试生产环境请使用Kafka、Pulsar这类消息队列作为Source否则作业重启后数据源就断了。3.2 Maven打包与依赖处理打包是整个提交流程里最能体现工程化水平的一步。很多人的作业在IDE里跑得好好的打成Jar提交到集群就报ClassNotFoundException十有八九是依赖没有打进去。这句话请记住Flink的核心依赖是集群自带的你的作业Jar里不要包含flink-streaming-java、flink-core这些Flink核心库否则会跟集群自身带的版本冲突。但第三方依赖比如连接器、JSON解析库、HTTP Client则必须打进Jar里或者放到Flink的lib目录下。推荐使用maven-shade-plugin来做打包它会自动把依赖合并成一个Fat Jar并且可以处理依赖冲突。build plugins plugin groupIdorg.apache.maven.plugins/groupId artifactIdmaven-shade-plugin/artifactId version3.4.1/version executions execution phasepackage/phase goals goalshade/goal /goals configuration transformers transformer implementationorg.apache.maven.plugins.shade.resource.ManifestResourceTransformer mainClasscom.example.SocketWordCount/mainClass /transformer /transformers filters filter artifact*:*/artifact excludes excludeMETA-INF/*.SF/exclude excludeMETA-INF/*.DSA/exclude excludeMETA-INF/*.RSA/exclude /excludes /filter /filters /configuration /execution /executions /plugin /plugins /build顺便说一句排除META-INF下的签名文件SF、DSA、RSA这个细节是很多人在打包环节踩坑的重灾区。如果不去排除某些依赖带签名信息Jar提交后可能出现SecurityException提示Invalid signature file digest。3.3 命令行提交与参数解析集群启动、作业打包完成之后就到了关键的提交环节。先启动一个Socket数据源用于测试# 在作业节点的终端上 nc -lk 9999然后执行提交命令# 在Flink安装目录下 bin/flink run \ -m jobmanager-host:8081 \ -c com.example.SocketWordCount \ -p 2 \ -d \ /path/to/your-job.jar一个一个参数来说。-m指定JobManager的地址和Web UI端口flink命令会把作业提交给JobManager的Dispatcher。如果不加这个参数flink命令会读取conf/flink-conf.yaml里的rest.address配置在本地模式下则默认提交到localhost:8081。-c指定包含了main方法的入口类。当你的Jar里有多个main方法时这个参数是必须的。虽然maven-shade-plugin在MANIFEST里写入了Main-Class但显式指定更稳妥也能避免别人拿错Jar包的时候一头雾水。-p是并行度这里指定2意味着作业的每个算子会启动两个并行实例。并行度设置的原则是不超过集群的总槽位数。比如你有两个TaskManager各4个槽位总槽位是8那么作业并行度最大可以设到8。超过的话作业会一直处于调度等待状态直到有槽位空出来。-d是detached模式提交后命令行立即返回作业在后台运行。如果不加这个参数命令行会一直跟着作业的运行状态作业结束命令才返回。实际生产环境推荐加-d把作业的运行交给集群管理。提交成功后命令行会返回一个Job ID同时提示作业已经提交到集群。你可以在Web UI的Job Manager页面看到作业由RUNNING到FINISHED的状态变化。如果你想用REST API来做提交Flink也提供了相应的接口。实际工作中很多自动化平台就是通过POST /jar/upload上传Jar再通过POST /jars/{id}/run来触发作业运行。这个方式比命令行更适合集成到调度系统里但细节比较多这里不展开。3.4 Web UI提交与SQL Client提交除了命令行Web UI也提供了直观的提交入口。在Web UI页面右侧有一个“Submit New Job”按钮点击后可以上传Jar包填写Main Class、并行度、Program Arguments等参数然后点击Submit提交。这种方式适合临时测试场景或者当你想快速观察作业在集群上的资源分配情况时使用。缺点是当你需要频繁提交、停止作业时手动点页面效率太低建议还是走命令行或者REST API。如果你的作业是基于Flink SQL开发的那就要用到SQL Client或者SQL Gateway。SQL Client是Flink自带的交互式SQL终端启动方式如下bin/sql-client.sh进入SQL Client之后你可以用标准的SQL语法来创建Source、Sink和计算逻辑比如CREATE TABLE source_table ( user_id STRING, page_id STRING, view_time TIMESTAMP(3), WATERMARK FOR view_time AS view_time - INTERVAL 5 SECOND ) WITH ( connector kafka, topic user_views, properties.bootstrap.servers localhost:9092, properties.group.id flink-sql-group, format json, scan.startup.mode latest-offset ); CREATE TABLE sink_table ( user_id STRING, cnt BIGINT ) WITH ( connector print ); INSERT INTO sink_table SELECT user_id, COUNT(*) AS cnt FROM source_table GROUP BY user_id;SQL Client的好处是开发效率高不需要写Java代码就能完成流式ETL缺点是不太适合复杂的业务逻辑和需要精细调优的场景。如果你想把SQL作业做成工程化、自动化的交付方式建议用SQL Gateway把SQL查询暴露成REST接口让上游调度系统直接调用。这两年Flink社区在SQL Gateway上投入很大如果你所在团队有多个业务线都要用Flink做实时计算SQL Gateway是个值得关注的方向。我在实际项目中还遇到过数据血缘的审计需求。用SQL方式开发时一张目标表的数据来自哪些源表、经过了哪些计算在SQL里天然是能追溯的但如果用DataStream API写这项能力就要自己实现。所以如果你们公司对数据治理有要求SQL优先是一个合理的选择。4. 常见问题与排查技巧实录4.1 资源不足与并行度设置作业卡在“ResourceManager请求Slot超时”这是我在群里被问得最多的问题。通常有两种情况一是你设置的并行度超过了集群可用槽位总数二是TaskManager内存分配不合理导致TaskManager启动后反复崩溃。先说第一种。假设集群有两个TaskManager每个4个槽位总共8个。你提交作业时并行度设成10那么有两个并行任务永远等不到槽位作业会一直处于SCHEDULED状态。解决办法很简单要么把并行度降下来要么在集群里增加TaskManager或槽位。第二种情况更隐蔽。TaskManager配置了2G内存但机器本身只有4G还同时跑了JobManager、Namenode、Kafka等一堆服务结果TaskManager进程一启动就被操作系统OOM杀掉。此时你去日志里看不到Flink报错只看到进程消失了。排查的时候一定要先看机器当前的可用内存。free -h如果内存确实比较紧张可以临时回收一些不需要的服务或者把TaskManager内存调低。但调低的时候要注意Flink分配给堆外内存managed memory、network memory是固定比例的你不能把总内存压得太小否则作业一跑就OOM。4.2 依赖冲突与ClassNotFoundClassNotFoundException是Flink作业提交后最经典的报错。我总结下来主要分两类。第一类作业代码里用了某个第三方库但打包时没打进去或者打进去了但版本不对。解决办法是检查Jar包。你可以用jar tf命令查看Jar内容确认对应类是否存在或者用jar -tf target/your-job.jar | grep 关键字来快速定位。第二类Flink核心库的版本冲突。比如你本地用Flink 1.17写的代码但集群是1.15提交后可能出现NoSuchMethodError。这种问题的根源在于Flink的核心类在集群端你的代码编译期用的版本和集群运行期加载的版本不一致。解决办法是保证开发环境和集群版本一致并在pom里把Flink依赖的scope设置为provided。dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version${flink.version}/version scopeprovided/scope /dependency还有一种情况是Jar包里出现了多个版本的同一个类。maven-shade-plugin在默认情况下不做依赖去重你需要通过relocation把有冲突的包重命名掉。这个技巧在处理guava、netty这类重依赖冲突时特别管用但配置比较复杂新手可以先避开等真遇到了再去研究。4.3 网络通信与端口问题StandAlone模式下JobManager和TaskManager之间的通信走的是Akka和Netty涉及多个端口。默认情况下JobManager的RPC端口是6123TaskManager的RPC端口是6122Blob服务端口是6124Web UI端口是8081。TaskManager启动时还会向JobManager注册注册用的就是RPC端口。我在实操时遇到过TaskManager日志一直报“Failed to connect to JobManager”的情况。检查下来发现是TaskManager节点上的防火墙把6123端口拦了。StandAlone模式下你不用每个端口都精确放行Flink支持配置端口范围比如taskmanager.rpc.port: 6122-6132这样TaskManager会在这个范围内选择可用的端口方便统一配置安全组。这个技巧在节点数量较多的时候特别有用省得一台一台去开放端口。4.4 任务运行中失败与恢复作业提交成功、运行一段时间后突然失败是另一种让人头疼的情况。如果是StandAlone模式下JobManager进程没有挂只是作业失败优先看JobManager日志和TaskManager日志。日志里如果出现“Checkpoint failed”字样说明检查点持久化有问题。StandAlone模式默认的检查点存储路径是JobManager的本地文件系统一旦JobManager重启检查点就丢了。生产环境建议把检查点配置到HDFS或者对象存储上state.backend: hashmap state.checkpoints.dir: hdfs:///flink/checkpoints state.savepoints.dir: hdfs:///flink/savepoints execution.checkpointing.interval: 60s execution.checkpointing.exactly-once: true这四行配置的含义分别是指定状态后端、检查点存储路径、Savepoint路径、检查点间隔和语义。其中exactly-once语义是Flink的核心卖点它保证故障恢复后数据不丢不重在特定条件下。配置完之后如果作业再次失败你可以在Web UI的Checkpoints页面看到最近一次Checkpoint的状态失败原因也会列出来排查起来会直观很多。5. 实操心得与优化建议5.1 我在实际项目中积累的经验做了一段时间Flink之后我有个体会提交方式本身不复杂复杂的是把整个流程想清楚。很多人在本地敲一行flink run就把作业提交上去了但没有想过这个作业需要多少个槽位、状态后端放在哪里、失败之后怎么恢复、日志怎么采集。我个人比较推荐的做法是这样的把提交命令固化成脚本或者接入调度平台在脚本里预设好并行度、内存参数、检查点路径、保存点路径最重要的是把Job ID记录下来。为什么要记录Job ID因为后续你想对作业做Stop、Cancel、从Savepoint恢复都需要这个ID。别等到要操作的时候再去Web UI里翻那效率太低了。还有一个容易被忽视的点提交作业的客户端机器必须要能访问到JobManager的REST端口但不需要能直接访问TaskManager。因为作业提交完成后Jar包是传到JobManager的Blob服务上再由JobManager分发给各个TaskManager的。你在远程客户端上提交时确保网络策略允许客户端访问8081和6124端口就行。5.2 从StandAlone到更复杂的部署形态如果你把StandAlone模式的流程完全搞懂了接下来可以尝试两个方向。第一个方向是把Flink作业容器化用Docker把JobManager和TaskManager包起来再结合容器编排平台做资源调度。第二个方向是提升作业的可观测性接入Prometheus监控指标、配置Flink的Metrics Reporter同时在作业里埋点记录处理延迟和数据量。我为什么建议先跑通StandAlone再说因为Flink的作业提交、状态管理、检查点恢复这些核心概念跟部署模式无关。你在StandAlone模式下把这些概念吃透了换到任何部署模式都只是换个环境变量和启动脚本的事。最后分享一个我自己一直在用的小技巧维护一个标准的作业部署清单。每次提交新作业之前按照清单检查一遍——并行度是否匹配集群槽位、检查点路径是否可写、日志级别是否合理、Jar包里的依赖是否干净。这个清单看起来简单但它帮我避免了至少五次线上事故。做实时计算的最怕的就是作业跑着跑着因为资源或者依赖问题挂了而你连它为什么会挂都不知道。把这些基本功做扎实比追求花哨的架构要实在得多。

Flink StandAlone模式作业提交全流程实战指南

相关推荐

CAD图纸坐标转GIS总对不上？椭球基准与七参数转换实战指南

C盘又红？从存储原理解析到系统盘空间清理与迁移实战

Event-Driven Architecture 事件驱动架构完整实践指南：Awesome Software Architecture 的事件驱动学习与实践路线

更多资讯

抖音音乐批量下载实战：douyin-downloader 配置与排错指南

GPU GEMM优化原理与实战：从Tensor Core到cuBLAS调优

gfast-ui v3.2 实战：Vue3+Vite+Pinia 后台开发与Nginx部署指南

一行脚本让网页拥有AI操作员：page-agent 项目终极入门指南

LEANN 贡献指南：从 uv 开发环境搭建到 PR 合入的完整工程实践

page-agent Chrome扩展完整指南：多标签页任务控制利器

今日更新

基于高阶循环谱的MPSK载波频率估计与MATLAB实现

MiniPdf方：.NET开源Office转PDF的高效解决方案

Solon AI v3.10：Java全栈智能体开发框架解析

本周精选

AI驱动的自动化专利分析平台技术解析

LeetCode-Go 题解：1252. Cells with Odd Values in a Matrix（奇数值单元格计数）双解法剖析

在线答疑系统Java毕设实战：Spring Boot前后端分离与状态流转

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？