做了几年实时计算我发现不少人把Flink当成一个“黑盒子”代码提交上去能跑就行至于任务调度逻辑、资源怎么分配全都交给框架。这种用法日常没问题可一旦遇到任务卡住、重启很慢、或者slot不够用就完全不知道从哪下手。这篇东西我想围绕Flink 1.12把任务调度逻辑讲透从你写的DataStream代码变成StreamGraph、JobGraph、ExecutionGraph的完整链路再到调度策略、Task Slot分配、故障恢复时的调度行为。适合刚开始接触Flink内部机制、或者已经在用1.12但想优化资源利用率的同学看完至少能知道作业从提交到运行每一步踩了什么油门、挂了什么挡。1. 调度链路全览一次作业提交背后的三层图流转1.1 StreamGraph、JobGraph、ExecutionGraph三个图分别管什么先建立整体认知。Flink里一个作业从用户代码到真正运行会经历三次“图”的转换每一层解决不同的问题。第一层是StreamGraph这是用户代码编译后的“逻辑DAG”。你在代码里addSource、map、keyBy、addSinkFlink会把这些算子节点以及它们之间的连接关系组织成StreamGraph。StreamNode就是算子节点StreamEdge是算子之间的数据通道。这一层很接近代码本身它不考虑多少台机器、多少并行度只记录“计算逻辑长什么样”。第二层是JobGraph这是客户端在提交给JobManager之前生成的“可提交作业图”。StreamGraph在JobGraph阶段会做一件非常重要的事Operator Chain合并。多个能串成一条线的算子会被合并成一个JobVertex多个JobVertex之间通过IntermediateDataSet产生数据交换。提交给JobManager的就是这个JobGraph。第三层是ExecutionGraph这是JobManager真正用来调度和执行的“并行展开图”。JobGraph里一个JobVertex只是一个逻辑顶点到了ExecutionGraph它会按照并行度展开成多个ExecutionVertex。比如某个JobVertex并行度是4那它就变成4个ExecutionVertex每个ExecutionVertex代表一个可以独立调度的并行子任务。顶点之间的边也会展开成ExecutionEdge代表实际的数据shuffle关系。等TaskManager真正接收任务后还会形成一张物理执行图一个Task线程对应一个ExecutionVertex中间通过网络连接完成上下游数据交换。通常我们说“调度”核心就是围绕ExecutionGraph展开的。理解了这三层图后面所有内容都不会跑偏。1.2 Operator Chain把多个算子打包成单个任务Operator Chain是调度逻辑里第一个影响资源形态的环节也是最容易被忽视的。很多同学看Web UI时发现“我明明写了source、map、sink三个算子怎么界面上只有一个Task”这就是因为Flink默认把可以合并的算子链成了一条线。Chain的合并条件在源码里有明确判断我总结下来主要是这几点上下游并行度必须相同上游到下游的分区方式必须是FORWARD也就是数据直接从当前算子流到下一个算子不需要经过keyBy、rebalance这类会改变分组和并行关系的操作下游算子最好只有一个输入union、多输入算子通常不适合链另外两个算子都没有被显式禁止chain。满足这些条件Flink就会把多个算子揉进同一个JobVertex最终在TaskManager上生成一个Task线程。这个机制对调度影响巨大。假设一个作业有5个算子并行度都是4如果不做chain那ExecutionGraph里就有20个ExecutionVertex需要调度做了chain并且合并成1个JobVertex后调度单元直接变成4个。调度器要处理的顶点少了五倍TaskManager之间也不会有无谓的网络传输。代价是任务粒度和故障恢复的边界变粗一旦某个子任务异常整条chain都要跟着重启。我自己的经验是默认chain不要轻易关除非某个算子需要独占线程做IO操作或者你希望某个高负载算子单独隔离否则关了chain只会白白增加调度和通信开销。1.3 JobGraph是怎么变成ExecutionGraph的JobGraph到ExecutionGraph的转换是JobManager收到作业后做的第一件大事。这个过程有个很直白的名字叫“并行化展开”。JobManager拿到JobGraph后遍历每个JobVertex根据并行度创建ExecutionJobVertex和它下面的多个ExecutionVertex。我画过一条线帮助自己记忆JobVertex是模板ExecutionJobVertex是这个模板在某个作业执行周期里的实例ExecutionVertex是实例里每个并行子任务的最小单位而Execution则是这个ExecutionVertex每一次具体的执行尝试。比如一个子任务失败后重新调度会生成新的Execution对象老的Execution会被标记成FAILED。每个Execution都有自己的状态机常见几个状态包括CREATED、SCHEDULED、DEPLOYING、RUNNING、FINISHED、CANCELED、FAILED。你之后看日志或Web UI时判断任务卡在哪一步其实就是看这个状态机停在了哪里。从调度框架的角度说JobGraph到ExecutionGraph转换完成之后接下来所有动作都集中在“调度ExecutionVertex”这件事上。这也正好引出Flink 1.12里最值得聊的调度框架重构。2. 1.12版本调度框架重构从逐任务调度到批量调度2.1 旧调度器的问题Flink 1.12之前的调度器核心逻辑大量集中在事件驱动和逐任务调度上。作业启动或者故障恢复时调度器会从图里挑一个ExecutionVertex先申请slot等slot分配到了再部署Task部署完成后再接着处理下一个。对于几十上百个并行度的作业这个过程就像排队买票一个一个来。这种方式的直接后果是故障恢复很慢。尤其是一个大作业在运行中挂掉你会看到任务不是一个一个起来的但也不是整齐批量起来的而是断断续续、像挤牙膏一样慢慢恢复。调度逻辑本身也不容易维护新的功能比如动态并行度、自适应调度想在旧框架上做改动非常吃力。流批一体推进之后批任务希望空闲资源来多少用多少、逐步拉起来流任务则希望尽可能快速恢复到稳定状态两种诉求差异越来越大旧调度器已经盖不住这栋楼了。2.2 SchedulingStrategy与EagerSchedulingStrategy1.12的最大变化是把“如何安排任务执行顺序”这摊事抽象成了SchedulingStrategy接口。调度器依然负责和资源管理器交互、部署任务但具体“先调度谁、后调度谁、一次调度多少”交给策略决定。默认的DefaultScheduler使用EagerSchedulingStrategy这是一次性调度策略。它在调度开始时会把当前可调度的所有ExecutionVertex全部提交调度请求批量申请slot拿到资源后并发部署。这种策略的好处非常明显作业启动不再有漫长的排队故障恢复时所有Task尽可能同时拉起耗时大大缩短调度逻辑也更简单少了各种状态回调和中间打断。除了EagerFlink还保留并实现了LazyFromSourcesSchedulingStrategy。这种策略更偏向批处理场景从Source节点开始一层一层往sink方向推进先把必要的最上游任务拉起来再根据结果逐步拉起下游。它适合资源弹性比较弱、希望作业边跑边获取资源的场景。1.12还新增了Adaptive Scheduler不过默认不启用这个我后面再说。这里必须提醒一点Eager策略不是一个“既要又要”的银弹。它要求集群有足够资源一次性满足整个作业的slot需求。在Standalone模式下如果资源不够作业不会像之前那样慢慢等到部分资源再启动剩余任务而是直接失败。这个坑后面排查部分会细讲。2.3 Slot共享与资源申请的核心逻辑说完策略再说调度器到底在申请什么资源。Flink的资源单位是Slot但这里的“Slot”不是简单的“一个Task占一个坑”这么粗暴。因为SlotSharingGroup的存在同一个共享组里的多个ExecutionVertex只要并行度相同是可以复用同一个Slot的。默认情况下所有JobVertex都会属于同一个共享组。这就意味着一个由source、map、window、sink串成且并行度一致的作业理想状态下所有Task都能被装进同一组Slot里。比如并行度为2那只需要2个Slot就能装下整条链上所有子任务每个Slot里的Task线程负责跑一条完整的计算链。这样做的好处是减少网络传输、提高线程利用率坏处是同一个Slot里的多个Task会抢同一个TaskManager的CPU和内存资源遇到一个算子特别耗CPU时会殃及同组其他任务。调度器申请Slot的内部流程也不复杂ExecutionVertex进入调度流程后会向Scheduler请求SlotScheduler通过SlotProvider向ResourceManager发起申请ResourceManager从当前可用的Slot池里分配一个Slot最后执行部署。这中间如果Slot不够在YARN模式可能会触发动态扩容等待但在Standalone模式往往就是失败或等待超时。理解这条链路后面排查“任务卡住”时会非常有底。3. 一次完整的调度部署过程从Job提交到Task运行3.1 JobManager、ResourceManager、TaskManager各自的调度职责很多新手分不清JobManager和ResourceManager到底谁负责什么。简单说JobManager是“调度的大脑”ResourceManager是“资源的仓库管理员”TaskManager是“干活的工人”。JobManager里运行着Scheduler和ExecutionGraph它决定一个ExecutionVertex应该什么时候调度、部署到哪个TaskManager。ResourceManager只负责Slot的分配和回收它不关心你跑的是什么算子只需要知道当前有多少空闲Slot、每个Slot属于哪个TaskManager。TaskManager启动时会向ResourceManager注册自己拥有的Slot之后接收JobManager的部署指令在本进程内创建Task线程来跑真正的用户代码。我用一个类比帮助理解JobManager是项目经理负责排工期、安排活ResourceManager是仓库管理员只管借出和回收工具TaskManager是工人工具到位了就干活。整个作业能不能跑起来其实是三方配合的结果。3.2 一个具体作业的调度走位示例光说概念还是虚我拿一个典型作业走一遍。假设代码结构是这样的DataStreamString stream env.addSource(new SocketTextStreamFunction(localhost, 9999)); stream .flatMap(new Tokenizer()).name(split) .keyBy(w - w.f0) .timeWindow(Time.minutes(5)) .apply(new MyWindowFunction()).name(count) .map(new FormatResult()).name(format) .addSink(new DiscardingSink()).name(sink);假设所有算子并行度都设置为2。由于flatMap与source之间是FORWARD关系并且并行度相同它们会chain成一个JobVertexkeyBy之后因为引入了key的shufflewindow和flatMap不可能chain但window到之后的map、sink如果也是FORWARD则它们可以继续chain。最终ExecutionGraph里可能有3个左右JobVertex并行度都是2所以一共产生6个左右的ExecutionVertex。默认SlotSharingGroup把所有JobVertex归到一个共享组并行度又一致那么调度器实际只需要申请2个Slot就能把这6个ExecutionVertex全部部署完成。更直观地说每个TaskManager上有一个Slot这个Slot里的Task线程会依次跑完整条计算链从source读到数据经过flatMap、window、map最后写向sink。现在反过来看如果我把sink的并行度单独调成4情况就变了。sink和前面的map并行度不一致无法chain在一起同时因为并行度不同默认共享组里同一个Slot无法再完整容纳“一个上游子任务一个下游子任务”的组合。调度器会让原有的2个Slot继续承载主链任务额外再分配2个Slot给sink的4个并行实例中的另外2个。原本2个Slot就能解决的作业现在需要4个Slot。调度逻辑的一举一动就这样直接影响资源瓶颈。3.3 从Web UI和日志观察调度结果作业跑起来之后我推荐大家养成看两个地方的习惯Web UI和JobManager日志。Web UI的Job Graphs页面能直接看到展开后的Task数量、每个Task的状态和所在TaskManager。如果一个Task一直显示SCHEDULED说明调度器已经决定要调度它但还没拿到Slot如果显示DEPLOYING说明Slot已经分配正在下发部署指令。JobManager日志里通常会有“Deploying task”和“Successfully registered task”这类记录可以从时间戳里推断调度耗时。我曾经排查过一个作业发现某些Task从SCHEDULED到RUNNING只用了3秒另一些却花了半分钟后来看日志发现是TaskManager分布在不同的物理机上网络拉取大状态时慢。这些观察说到底都是为了让调度过程“可见”而不是遇到了再说。4. 常见问题排查与故障恢复4.1 任务一直SCHEDULED怎么定位任务卡在SCHEDULED状态是我收到最多的求助类型之一。先说结论绝大多数情况都是Slot申请不到。第一步去JobManager日志里搜Could not allocate或者Not enough resources如果看到了类似错误那说明调度器正在等ResourceManager给它一个可用的Slot。第二步去ResourceManager或TaskManager列表页看当前空闲Slot数量。如果空闲Slot是0而任务又在等Slot那原因基本锁定集群资源不足。但还有一种容易忽略的情况资源足够但因为Slot共享组设置不当导致部分Slot被算子的某个并行实例“独占”了后续Task找不到可匹配的共享Slot。这种时候日志不一定直接报资源不足更可能出现反复重试申请。我建议这时候把作业的并行度、共享组、chain关系列出来手工算一遍究竟需要几个Slot再对比集群实际的Slot数。如果任务卡在DEPLOYING那又是另一条排查路径通常是TaskManager和JobManager之间的RPC连接有问题或者TaskManager本地状态加载太慢。DEPLAYING卡住很少见但一旦出现先检查网络和TaskManager心跳。4.2 Standalone集群slot不足时的Eager调度失败案例我实际踩过一个很典型的坑。某个作业逻辑不复杂主链并行度是3sink并行度是6集群是4个Slot的Standalone模式。我当时粗略一算觉得3个Slot肯定够结果作业一提交就失败。后来把ExecutionGraph里的ExecutionVertex数拉出来看主链某JobVertex有3个子任务sink有6个子任务。因为并行度不一致同一Slot里不能把主链子任务和sink子任务完全配对实际需要4到6个Slot。Eager调度器一次性申请了超过4个的资源直接判定失败。加上我配了固定延迟重启策略这个作业就在“失败-重试-再失败”里打转日志看起来像死循环其实是每一步部署连第一步都跨不过去。这个案例给我两个教训第一估算Slot需求不能只看总并行度要结合chain和SlotSharingGroup算第二在Standalone等无动态扩容的环境里Eager策略对资源预估的要求比老版本更严格资源不足时会很快暴露而不是给你等待的机会。解决方式也简单要么把sink并行度调回与主链一致要么提前把集群Slot数提上去要么换到能动态申请资源的部署模式。4.3 failover时调度器在做什么作业运行中Task挂掉之后调度器不是简单地把同一个Task重新部署一次。它会触发一个完整的Local Recovery流程先确认受影响的ExecutionGraph区域取消这些区域内的Task释放它们的Slot然后根据用户配置的RestartStrategy决定是否重试、重试几次。Flink 1.12默认的故障恢复策略是Region-based Failover也就是按“流水线区域”隔离故障。一个Task失败后只会重启与它直接连通的pipelined区域而不是全部作业都重启。这对长链路作业来说非常重要能大幅减少恢复范围。调度器在恢复时依然走Eager策略把受影响区域的全部ExecutionVertex一次性调度起来。所以你在日志里会看到这样一个模式区域内的Task几乎同时进入DEPLOYING然后一起RUNNING。相比老版本“逐个重启”这个恢复速度体感明显更快。如果想切换策略可以通过jobmanager.execution.failover-strategy配置可选full或region生产环境我建议保留region。同时要注意恢复时间不只是调度时间还包括从Checkpoint或Savepoint加载状态的时间状态大时即使调度再快整体恢复也快不起来。5. 基于调度逻辑的调优经验5.1 并行度与slot数怎么设计理解了调度逻辑之后设计并行度和Slot数就不再是拍脑袋了。我的经验是先做减法再做加法。先确定瓶颈算子的并行度比如窗口聚合对CPU和内存要求高那就以它为基准然后让主链路算子的并行度尽量一致这样Operator Chain能最大化合并Slot共享也能把资源压缩到最低。常规做法是让一个TaskManager的Slot数等于它的CPU核心数并行度整体接近集群总核心数这样线程切换最少、数据本地性也好。对于明显的高负载算子比如复杂的自定义函数或大状态窗口我建议不要让它和普通算子挤在同一个默认共享组里。Flink支持通过slotSharingGroup为算子指定独立共享组把高负载算子隔离到专属Slot避免它把同组其他任务拖死。但也要注意组一旦分开Slot总数很可能会上升需要重新评估集群容量。5.2 关键配置项速查下面这些配置是我在1.12上最常用的调度相关参数整理成一张速查表配置项作用我的推荐parallelism.default默认并行度按集群规模和瓶颈算子的需求设置taskmanager.numberOfTaskSlots每个TaskManager的Slot数一般等于CPU核数不要贪多jobmanager.scheduler调度器实现默认Default生产先别动Adaptiveslotmanager.processing.timeoutSlot申请超时时间集群排队场景可适当调大restart-strategy失败重启策略流作业用fixed-delay重试次数别太高jobmanager.execution.failover-strategy故障恢复范围默认region保持默认即可单独提醒一下slotmanager.processing.timeout这个值默认是10秒左右Eager调度一次性申请大量Slot时如果资源不足需要等待超时时间太短会很快报错。在YARN等能动态扩容的环境中可以谨慎调大一些给资源申请留出缓冲。5.3 我踩过的几个坑最后分享几个真实经历都是调度逻辑在实操里最容易绊倒人的地方。第一个坑就是升级到1.12之后我一度觉得“重启变快了”是全自动的其实完全依赖新调度器和Eager策略。如果某个作业显式关闭了chain、并行度参差不齐调度负担依然会很大启动慢的问题也不会消失。第二个坑Web UI上Task变少了不是调度器漏调度。默认chain开启时几个算子合成一个Task是正常现象不要被“少得可怜”的任务数吓到。想验证是否正确可以在作业里调用disableOperatorChaining()对比一下启动时间和资源消耗立刻能感受到chain对调度的影响。第三个坑sink或source的并行度和主链不一致会让Slot占用翻倍。我前面那个sink并行度从3调到6导致作业失败的案例就是典型代表。现在我做并行度调整一定会先动手算一遍最终Slot需求再改。第四个坑暂时不要在生产环境尝试1.12里的Adaptive Scheduler。虽然这个概念很吸引人但1.12里它还属于实验特性默认没有启用。我建议先在测试环境跑通确认它的资源调整行为和你的部署模式匹配再考虑上生产。等后续版本成熟了这类调度器才有机会成为默认选项。回过头看我自己的项目经历有一次凌晨线上作业挂了Eager调度器把所有Task几乎同时拉起配合region failover和checkpoint恢复整个作业十来分钟就回到了稳定状态。而同样规模的老版本作业可能还在逐个Task排队恢复。调度逻辑这东西平时藏在框架底层看不见但一到故障和资源紧张的时候它就会用结果告诉你设计得好不好差别是天上地下。你们如果也在折腾Flink 1.12我建议多花点时间把JobManager日志里slot申请记录和Web UI上的Task部署时间线拉出来看一遍那比对着文档背概念更能把调度逻辑真正落到脑子里。
