企业官网建设流程全解析

先说点实在的，把 Flink 1.12 的任务调度逻辑讲清楚，不是背官方文档，也不是贴几张架构图就完事。很多人看 Flink 源码或者调作业的时候，经常遇到一个现象：作业提交了，JobManager 也收到了，但 Task 就是迟迟不启动，或者明明有空闲 Slot，作业却一直卡在 RUNNING 状态不动。这些问题说到底，都是没把调度逻辑吃透。我基于 1.12 版本源码和线上实际调优经验，把这块从头到尾梳理一遍，该说原理的时候说原理，该给结论的时候给结论。

1. 先搞懂调度到底“调”的是什么：JobGraph 到 ExecutionGraph 的桥梁作用

Flink 的任务调度，很多人第一反应是“把 Task 分配到 Slot 上”。这句话没错，但它太笼统了。真正要理解 1.12 的调度逻辑，必须从数据结构层面讲起——调度调度的本质，其实是两套图的转换和状态流转。

1.1 StreamGraph、JobGraph、ExecutionGraph 三者关系

你在 Flink 里写的 DataStream API 或者 Table API 程序，翻译过来其实要经过三层图结构：

• StreamGraph：用户代码直接生成的原始图，由 StreamNode 和 StreamEdge 组成，它是最贴近业务逻辑的一层，包含了算子之间的上下游关系。
• JobGraph：StreamGraph 经过优化（比如链化 operator chain）后生成的中间表示。这一层已经合并了可以串联在一起的算子，是提交给 JobManager 的实际单位。JobGraph 里的顶点叫 JobVertex，边叫 JobEdge。注意，JobVertex 携带了并行度、Slot 共享组、算子链信息等关键调度属性。
• ExecutionGraph：JobManager 收到 JobGraph 后，为每个 JobVertex 创建对应的 ExecutionVertex。ExecutionVertex 是并行子任务的调度单元，每个 ExecutionVertex 又可以对应多次 ExecutionAttempt（重试、故障恢复都会生成新的 attempt）。这一层才是真正执行调度的图结构。

很多人容易混淆的一点是：调度器盯着的不是 JobGraph，而是 ExecutionGraph。JobGraph 是“静态蓝图”，ExecutionGraph 是“动态运行实例”。所有 Slot 分配、任务部署、状态切换，全部作用在 ExecutionVertex 这个级别。

1.2 调度信息从哪里来：JobInformation 与 TaskInformation

继续往深挖一层。JobManager 在接收到 JobGraph 之后，会为每个 JobVertex 生成 TaskInformation——里面包含算子名称、并行度、最大并行度、每个子任务的索引范围。这些信息打包进 Task 部署的 TaskDeploymentDescriptor，也就是 TaskManager 最终收到的“任务任命书”。

这块的关键点在于：调度不是碰运气找 Slot，而是带着约束条件去找 Slot。每个 ExecutionVertex 其实带着一堆调度约束：要几个 Slot、Slot 属于哪个 TaskManager、要不要和其他 Task 共享 Slot、有没有 Location 约束（比如延迟调度场景下，State 在某个 TaskManager 上，就希望调度到那个节点附近）。理解了这一点，再去看为什么有些 Task 就是调度不动，心里就有数了——多半是约束条件无法满足，而不是集群资源真的不够。

1.3 调度决策的发生时机

1.12 版本在调度时机上有过一次“分水岭式”的改动：从 Eager Scheduling 到 Pipelined Region Scheduling。这直接影响作业启动效率、资源利用率和故障恢复成本。关于这个我会在第三部分专门展开，这里先记住一个结论：调度时机本质上取决于作业内部是否存在阻断边（Blocking Edge）。如果作业全部是流水线边（Pipelined Edge），理论上可以等所有 Task 一起启动；如果有阻断边（比如 Reduce 前的 Shuffle、迭代边界），则需要按 Region 分阶段调度。

2. 调度器的“内核引擎”：ExecutionGraph 的状态机与调度策略

解决了“调什么”之后，要回答“怎么调”。1.12 版本的核心调度器是 DefaultScheduler，它统一处理任务状态转换、部署请求、故障恢复和资源申请。这一节我们拆开它的内部逻辑。

2.1 四个核心状态：CREATED / SCHEDULED / DEPLOYING / RUNNING

每个 ExecutionAttempt 都有一条状态机路径：

• CREATED：ExecutionVertex 被创建，但还没向 ResourceManager 申请资源。
• SCHEDULED：已经拿到 Slot，ExecutionAttempt 与 Slot 完成绑定（包含 Slot 的物理地址、资源 ID）。
• DEPLOYING：JobManager 把 TaskDeploymentDescriptor 发给 TaskManager，等待 TaskManager 确认接收并启动。
• RUNNING：TaskManager 上报任务真正启动成功。

这个状态机在外层还叠了一层 ExecutionGraph 的状态（RUNNING / FAILING / RESTARTING / CANCELLING 等）。排查问题的时候，先看作业整体状态，再看每个 Task 处于哪个子状态，基本能定位到瓶颈在哪一环。

2.2 DefaultScheduler 的调度主流程

DefaultScheduler 的核心逻辑可以概括成三步循环：

1. 收集待调度 ExecutionVertex。调度器维护了一个“待调度顶点集合”，初始时是所有没有依赖或依赖已满足的 ExecutionVertex。对于流水线边连接的顶点，通常需要等待整个 Region 可调度。
2. 申请资源并分配 Slot。调度器根据顶点资源需求（TaskManager 的 slot 资源模板，默认是 1 个 slot 对应 1 个 Task）向 ResourceManager 发起申请。注意这里不是每个 Task 单独申请，而是按 Slot 资源粒度申请。
3. 部署 Task。拿到 Slot 后，调度器把 TaskDeploymentDescriptor 投递给对应 TaskManager，然后监听部署结果。

一个容易忽略的细节是：调度器是异步的。资源申请、Task 部署都是异步消息交互，所以调试时看到某个 Task “卡住”，不是真的死锁，而是它在等某个异步回调返回。这时候要去看 JobManager 日志里有没有资源申请超时或者 Task 部署超时的报错。

2.3 Slot 分配策略：从“随便找个空位”到“带约束的资源匹配”

Slot 是资源调度的基本单位。1.12 版本 TaskManager 的 Slot 默认大小是 1 个 Task（对应 1 个 CPU 核心和固定内存）。但真正分配 Slot 的时候，逻辑比这复杂得多：

• Slot 是否可用：不是看 Slot 是否为空，还要看它的状态是 ACTIVE 还是 RELEASED。如果 TaskManager 正在释放资源，Slot 暂时不可分配。
• Slot 是否满足 Location 约束：如果任务有状态需要从上一次 failover 恢复，并且状态存储在某个 TaskManager 的本地磁盘，那么调度器会优先把任务调度回那个 TaskManager（Locality 优先）。1.12 里是通过 StateLocation 记录来实现的。
• Slot 是否满足 Slot 共享组约束：Flink 允许不同算子的子任务共享同一个 Slot（SlotSharingGroup），这能显著降低资源碎片。共享组相同且没有冲突的情况下，多个 ExecutionVertex 可以绑定到同一个 Task 上运行。

这里给大家一个实操经验：如果你发现 Task 调度不上去，先不要急着加机器。用flink list看作业状态，再用 JobManager 日志看具体是“No resources available”还是“Not enough resources to satisfy all tasks”。如果是后者，往往是因为你的 Slot 共享组配置不合理，导致明明有空闲资源却用不上。

3. 1.12 的“重头戏”：Pipelined Region Scheduling 是怎么改变调度时机的

3.1 为什么 Eager Scheduling 不再合适

在 Flink 1.12 之前，默认调度策略是 Eager Scheduling：作业的所有 ExecutionVertex 一次性全部申请资源、全部部署。这种策略的好处是逻辑简单，状态一致性好（要么全部起来，要么全部失败）。但代价也很明显：

• 资源申请峰值高。一个作业如果有 100 个 Task，Eager 调度会同时申请 100 个 Slot，即使某些 Task 完全可以等数据到达后再启动（比如 Reduce 算子要等 Map 阶段产生数据）。
• 启动门槛高。资源不足时，哪怕只有 1 个 Task 申请不到 Slot，整个作业都无法启动。这对于大作业、共享集群来说非常不友好。
• 故障恢复被放大。某个 Task 失败需要重启时，Eager 策略往往要把整个作业拉起来重启，而不是局部恢复。

3.2 Region 的划分逻辑：阻断边是分水岭

Pipelined Region Scheduling 的核心思想是：把 ExecutionGraph 按阻断边切割成若干个 Region，只有上游 Region 全部执行完成后，下游 Region 才开始调度。而 Region 内部的流水线边连接的所有顶点，会作为一个整体一次性调度。

这里必须解释清楚“阻断边”和“流水线边”的区别：

• 流水线边：上下游之间是流式传输，上游产生一条记录就能立刻发给下游处理。比如 map → filter 之间的边。
• 阻断边：上下游之间存在“屏障”，上游必须全部执行完，才能把完整结果交给下游。典型例子是 Batch 模式下的 Shuffle、迭代计算边界，以及某些需要全局排序的场景。

1.12 的调度器会遍历 ExecutionGraph，把所有由流水线边连通的顶点聚成一个 Region，然后在阻断边处断开。调度时，Region 之间按拓扑序依次调度，Region 内部则整体推进。

3.3 Region 调度带来的实际收益与代价

我用一个线上案例来说明。某个离线批作业是两层结构：第一层是数据读取和预处理，共 20 个 Task；第二层是聚合计算，共 5 个 Task。Eager 调度下，集群必须同时有 25 个空闲 Slot 才能把作业提交；Region 调度下，第一层 Region（20 个 Task）先启动，等第一层全部完成后，第二层 Region（5 个 Task）才申请资源。

收益显而易见：资源申请峰值从 25 降到 20（如果第一层先跑完再申请第二层，峰值甚至可以更低）。而且如果第二层 Region 资源不足，不影响第一层执行。代价则是：作业总执行时间可能变长，因为 Region 之间是严格串行的，下游不能提前“偷跑”。

这里要吐槽一下：很多人一看 Region 调度就以为能自动解决所有资源问题，其实不然。如果你把整个作业写成一个巨大的流水线（全部是流水线边），那 Region 调度退化成 Eager 调度，没有任何收益。所以写作业时要有意识地在合适的地方引入阻断边（最典型的就是 Batch 模式下的 shuffle + 聚合），让调度器有切分的空间。

3.4 如何确认当前作业触发了 Region 调度

想验证自己的作业是否有阻断边、是否走了 Region 调度，最简单的办法是看 JobManager 日志。作业启动时，日志里会打印类似“Scheduling the execution ... in region”这类信息。另外可以通过 Web UI 的任务列表观察：如果某些 Task 在图里的启动时间明显晚于上游 Task，而不是同时启动，说明走了 Region 分阶段调度。

4. 从调度到部署：Slot 申请、TaskDeploymentDescriptor 与 TaskManager 的执行链路

调度器做决策只是上半场，下半场是真正把任务扔到 TaskManager 上跑起来。这段链路经常出问题，而且问题往往不体现在调度逻辑本身，而是体现在消息传递和资源管理上。

4.1 Slot 申请是怎么从调度器走到 ResourceManager 的

调度器确定某个 ExecutionVertex 需要 Slot 后，会调用 ResourceManager 的申请接口，传递的信息包括：

• Slot 请求的资源画像：默认是 1 个 Slot 对应 1 个 Task，即一个 TaskManager 上有几个 Slot，就能并行跑几个任务。
• JobMaster 的定位信息：TaskManager 分配好 Slot 后，需要能回连到 JobManager（实际是 JobMaster）来进行任务部署。
• 申请超时时间：默认是 5 分钟，超时会触发重新申请或者失败。

ResourceManager 收到请求后，会先在当前已知的 TaskManager 注册信息里找空闲 Slot。找不到时，会向资源提供商（Standalone 模式下是手动启动的 TaskManager，YARN/K8s 模式下是资源管理框架）发起扩容请求。

这里有一个重要细节：Slot 的分配是“预占”还是“实占”。在 Flink 1.12 中，Slot 一旦被分配给某个作业，就进入 ALLOCATED 状态，哪怕作业还没真正把任务部署上去，其他作业也不能占用这个 Slot。所以你会看到集群明明有 Slot，但其他作业就是等不到资源——因为 Slot 被一个还没跑完的作业占着。这种现象在大作业多作业混合部署时很常见，解决办法是调低 Slot 空闲释放时间，或者控制最大并行度。

4.2 TaskDeploymentDescriptor 里到底装了什么

Slot 分配成功只是第一步。接下来 JobManager 会把任务的所有必要信息打包进 TaskDeploymentDescriptor，主要包括：

• JobID 和 JobInformation：任务归属哪个作业。
• TaskInformation：任务所属的 JobVertex、并行子任务索引、算子链信息。
• 执行配置：包括重启策略、状态后端配置、Checkpoint 配置等。
• Slot 信息：任务被分配到的 Slot 编号、物理地址。
• 用户代码类加载信息：1.12 里支持动态用户代码加载，TaskManager 可能需要从远程拉取用户 JAR。

这个描述符会被序列化后发给 TaskManager。TaskManager 收到后，反序列化、创建 Task 对象、启动线程，整个过程完全异步。

4.3 TaskManager 侧的三板斧：创建、注册、心跳

TaskManager 收到 Task 部署请求后的处理流程可以概括为：

1. 创建 Task 对象：从 TaskDeploymentDescriptor 中反序列化出 TaskInformation，然后给任务分配内存、创建输入输出 gate。
2. 注册到 TaskExecutor：TaskExecutor 是 TaskManager 内部管理多个 Task 的服务，它负责维护 Slot 状态和 Task 生命周期。
3. 上报启动结果：如果一切正常，TaskManager 向 JobManager 发送 Task 启动成功消息；如果失败，发送失败消息并附上异常堆栈。

还有一点：心跳机制对调度的隐性影响。在 Flink 1.12，JobManager 和 TaskManager 之间有心跳超时检测。如果 TaskManager 因为负载过高导致心跳超时，JobManager 会认为该 TaskManager 失效，进而触发资源释放和任务重启。很多线上“莫名其妙失败”的作业，其实是 TaskManager 失联导致的。

4.4 任务部署失败时的调度器行为

任务部署失败（比如 Task 初始化抛异常、TaskManager 进程崩溃），调度器会走 failover 流程。1.12 默认是 RestartAllStrategy（重启所有任务），Region 调度下也可以配置成只重启受影响 Region。这里需要特别注意：failover 会重新触发调度，也就是说，所有 Slot 可能被释放，然后重新申请、重新分配。如果你的集群资源紧张，failover 会导致“雪崩效应”：一个 Task 失败，全部重启，重启时资源不够，作业卡死。

我的实战建议是：对于关键批处理作业，配置好重启次数和延迟间隔（flink-conf.yaml 里的 restart-strategy.fixed-delay 系列参数），并且确保集群有一定的 Buffer 资源，不要精确跑满，否则 failover 基本等于作业终结。

5. 调度相关问题的完整排查链路：从“作业卡住”到“找到根因”

最后一部分，用我在实际运维中遇到过的三类典型问题，把排查思路完整过一遍。这些问题每个都踩过坑，按这个思路走能少走很多弯路。

5.1 现象一：作业提交后一直 RUNNING，但 Task 数远小于并行度

典型场景：作业申请了 50 个 Slot，但实际只有 20 个 Task 在跑，其他 Task 一直处于 SCHEDULED 或 CREATED 状态。

排查链路：

1. 打开 Web UI 看 Task 状态，先确认卡在哪个状态。如果大量 Task 停在 SCHEDULED，说明 Slot 申请阶段就在等待。
2. 去 JobManager 日志搜关键字“Not enough resources”或“No resources available”。注意区分：前者是说集群没有满足条件的 Slot；后者是说集群完全没有可用 Slot。
3. 看 ResourceManager 日志，确认它是否收到了 Slot 申请请求，以及它内部维护的空闲 Slot 列表。
4. 用curl访问 JobManager 的 REST API，查看 TaskManager 和 Slot 使用情况，确认是否有些 TaskManager 已经失联（Timeout）。

根因：大部分情况下是集群资源碎片化，比如多个 TaskManager 各自只有少量空闲 Slot，但作业需要的是大量连续 Slot（且因为 Location 约束，只认特定节点）。

排解手段：如果是 Standalone 模式，考虑把 TaskManager 的 Slot 数调到合理值（避免 slot 过小导致碎片）；如果是 YARN 模式，检查容器资源申请是否与 TaskManager 内存配置匹配。

5.2 现象二：作业启动后某几个 Task 反复失败重启，整体呈现“震荡”

典型场景：一个作业持续运行，但某个 Task 每隔几分钟就挂一次，然后整个作业重启，导致 Checkpoint 一直失败，吞吐骤降。

排查链路：

1. 先看失败 Task 的异常堆栈，确定是任务自身 bug 还是资源问题。
2. 观察失败时间点是否与内存增长曲线吻合。如果是明显的 OOM（OutOfMemoryError），去查 TaskManager 日志确认堆内存使用情况。
3. 确认失败是否与 GC 停顿有关：如果 Full GC 时间过长，可能导致心跳超时，JobManager 判定任务失联。
4. 检查 Checkpoint 是否成功完成。如果某个 Task 的 Checkpoint 一直超时，会影响整个作业的恢复能力。

根因：我在实际处理中遇到过三种情况：一是 TaskManager 堆内存设置过小，被大状态撑爆；二是网络反压导致 Task 长时间阻塞，被误判为失联；三是第三方依赖包在部分 JVM 环境下触发死锁。

排解手段：升级依赖版本；调整 TaskManager 内存分配，把托管内存（Managed Memory）和堆内存比例调优；开启网络反压告警，提前干预。

5.3 现象三：Slot 空着，但新作业怎么也申请不到资源

典型场景：集群有 5 个 TaskManager，每个 8 个 Slot，一共40个。上面只跑了一个占用 10 个 Slot 的作业，其余 30 个 Slot 空闲，但提交一个新作业却报资源不足。

排查链路：

1. 先确认空闲 Slot 是否真的进入“可分配”状态。Flink 的 Slot 从“Task 结束后”到“释放完成”有个异步过程，可能需要几十秒。
2. 查看新作业的 Slot 请求是否带了特殊约束，比如它要求所有 Task 必须落在同一个 TaskManager 上（某些自定义 SlotSharingGroup 配置会导致这种约束）。
3. 检查集群是否配置了 Slot 资源模型，比如 TaskManager 的 Slot 数是否配对了 CPU 和内存。如果资源配置不合理，Slot 可能“看得到吃不到”。
4. 看 ResourceManager 的日志，确认空余 Slot 是否被标记为 reserved（比如用于 failover 的预留）。

根因：最常见的原因是 Slot 共享组配置不一致，导致新作业要求独占某些 TaskManager，但这些 TaskManager 上还残留着旧作业的 Slot 足迹（即使任务已经结束，Slot 还挂在作业的分配记录里，需要等待释放周期）。

排解手段：调短 Slot 释放周期（slot.idle.timeout）；为新作业显式配置 SlotSharingGroup，确保它可以在任意 TaskManager 上调度；如果条件允许，给作业单独分配集群环境。

5.4 一个通用建议：把调度的“黑盒”变“白盒”

Flink 1.12 的 Web UI 其实已经提供了一部分调度信息，比如每个 Task 的起始时间、状态、所在 TaskManager 地址。但更细粒度的调度过程，建议开启调试级别日志：

logging: loggers: org.apache.flink.runtime.jobmanager: DEBUG org.apache.flink.runtime.resourcemanager: DEBUG org.apache.flink.runtime.taskexecutor: DEBUG

开启后，你会在日志里看到每一步 Slot 申请、分配、部署的详细记录。排查问题时按时间线串联这些日志，基本能还原完整的调度链路。我习惯的做法是：把 ResourceManager 的日志单独输出到一个文件，这样排查作业调度问题时不用翻整个 JobManager 的日志，效率能提升不少。

6. 调度参数的“低成本优化清单”：不写代码也能提升调度效率

如果你不想深入源码，也想让调度更顺畅，有几个参数值得重点关注。这些参数我在多个生产环境调过，效果非常直接。

TaskManager 内存模型相关

• taskmanager.memory.process.size：TaskManager 总内存。不要拍脑袋定，要算：堆内存 + 托管内存 + 直接内存 + JVM 开销。
• taskmanager.memory.managed.fraction：托管内存占比，默认 0.4。如果你的作业大量使用 RocksDB 状态后端，可以适当调高；如果主要用堆内存状态，可以调低。

Slot 资源相关

• taskmanager.numberOfTaskSlots：每个 TaskManager 的 Slot 数。建议不要设得太大，否则单机资源碎片严重，调度器找不到满足条件的 Slot 的概率会上升。一般 4~8 比较稳妥。
• parallelism.default：默认并行度，影响 Slot 申请的批大小。

调度超时相关

• slot.request.timeout：单次 Slot 申请的等待时间，默认 5 分钟。资源紧张时，如果业务允许，可以适当调大，避免频繁超时重试。
• akka.ask.timeout：消息交互超时，影响调度器与 TaskManager 之间的任务部署确认。如果集群网络波动较大，可以适当调大，但不要过于激进，否则故障发现会很慢。

故障恢复相关

• restart-strategy.fixed-delay.attempts：失败重试次数。
• restart-strategy.fixed-delay.delay：两次重启之间的延迟，建议给几秒到几十秒，给集群资源释放和重新分配留出时间。

有一点必须提醒：没有一套参数能适配所有作业。我见过有人把 Slot 数调到 32，结果作业确实并行上去了，但单个 TaskManager 内存被耗尽，频繁 OOM。调参一定要结合自己的任务类型、数据量、状态大小来定，最好通过压测找出临界点，再上生产。

7. 关于调度逻辑的一句话心得

从 Flink 1.12 开始，调度已经不再是一个“启动时全量拉起的硬逻辑”，而是变成了“按需、分阶段、带约束”的资源匹配过程。很多人只盯着并行度和 Slot 数量，实际上调度的约束条件才是决定作业能否跑起来的隐形推手。多花点时间理解 Region、SlotSharingGroup、资源申请超时这些概念，会比盲目堆机器有效得多。

最后补一个我自己常用来快速判断作业调度状态的方法：提交作业后，盯着 Web UI 看 Task 启动的时间戳分布。如果所有 Task 几乎同时出现，那是 Eager 式调度；如果 Task 分批次出现，并且符合作业的 Region 拓扑顺序，说明分阶段调度生效了。这个观察方法不需要看日志，几秒钟就能得出初步结论，排查问题时很省事。