把 TaskManager、Slot 和 Parallelism 放在一起讲是因为我见过不少同行把这三个词混着用最后在处理延迟和资源利用率问题上绕了大弯。Parallelism 决定一个算子要起多少个并行实例Slot 决定 TaskManager 能为这些实例提供多少个资源位置而 TaskManager 的数量乘以 Slot 数量又共同决定了整个集群能承载的并行规模。搞不清这三者是怎么协作的就会出现“明明设置了并行度 16运行起来却只有 8 个任务”“集群堆了一堆 TaskManager作业却迟迟调度不起来”这种看起来很反常的现象。这篇文章不打算复读官方文档而是结合我自己的排障和调优经验把这三个概念的定位、换算方法和实践教训一次说清楚。1. 理解 TaskManager、Slot、Parallelism 之前先看清一次任务调度全貌1.1 提交一个 Flink 作业内部究竟发生了什么很多同学把注意力放在 API 写法上忽略了作业提交之后的那条链路。实际上从你点击提交到任务真正开始计算中间要经历好几层图的转换。你写的DataStreamAPI 代码会先生成StreamGraph这是一个不包含并行展开的逻辑图。然后客户端会把相邻且满足条件的算子合并成算子链生成JobGraph。这个 JobGraph 是给 JobManager 看的作业蓝图。JobManager 收到之后再把它展开成ExecutionGraph这时候每个 JobVertex 才真正变成多个并行子任务ExecutionVertex。接下来是调度器的工作它会为每个需要执行的子任务寻找一个空闲的 TaskManager Slot找到之后把任务下发到对应 TaskManager 的线程池中执行。TaskManager 是真正跑用户代码的地方JobManager 只负责协调、调度、检查点等管理工作不参与具体计算。所以这条链路可以简化为用户代码 → StreamGraph → JobGraph → ExecutionGraph → Task 被调度到 TaskManager 的 Slot 中执行。Parallelism 决定的是 ExecutionGraph 里一个顶点有多少个并行子任务Slot 决定的是 TaskManager 能同时容纳多少个这样的任务而 TaskManager 的总数和 Slot 的总数一起决定了一个集群在当前时刻最多能铺开多大的执行规模。1.2 TaskManager 是进程Task 是执行单元Slot 是资源容器我经常在答疑群里看到这样的说法“我启动了三个 TaskManager并行度设了 6为什么只有 3 个任务在跑”这里混淆的其实是三个层面的东西。TaskManager是一个 JVM 进程它在 Flink 集群中负责执行任务。一个物理机或容器里可以跑一个 TaskManager也可以跑多个但生产中通常一个容器一个进程方便资源隔离和故障恢复。一个Task是 Flink 执行引擎的最小执行单元。它可能对应一个被合并后的算子链也可能对应一个单独的算子。TaskManager 内部用线程池来跑这些 Task每个 Task 会在自己的线程里执行用户逻辑。一个Task被并行化后会有多个实例每个实例叫 SubTask。比如某个算子的并行度是 4那么它的逻辑会被拆成 4 个 SubTask每个 SubTask 处理一部分数据。Slot则是 TaskManager 内部的资源容器。TaskManager 启动时就会宣布自己有几个 SlotJobManager 只能把任务调度到空闲的 Slot 上。Slot 可以理解为“资源位置”它并不直接等于线程数也不直接等于 CPU 核数但它决定了这个 TaskManager 最多能同时接收多少个任务。这三者的关系可以这样类比TaskManager 是一栋办公楼Slot 是楼里的办公室Task 是被安排的团队。Parallelism 决定你要组建多少个团队而能不能同时开那么多团队取决于整栋楼里有多少间办公室。1.3 调度器按 Slot 分配任务而不是按 CPU 核数这一点是新手最容易忽略的Flink 的调度器不关心你的机器有几个核它只关心 Slot 空闲还是不空闲。假设你的 TaskManager 机器是 8 核但你把taskmanager.numberOfTaskSlots配成了 2那么 JobManager 认为这个 TaskManager 最多只能接收 2 个任务。即使机器上还有大量 CPU 闲着作业也会因为没有空闲 Slot 而排队等待。反过来如果你把 Slot 数配成 16而机器只有 8 核JobManager 会认为这个 TaskManager 能接收 16 个任务于是同时把 16 个 Task 调度过去。结果就是 8 个核要被 16 个任务线程抢导致线程上下文切换频繁吞吐反而不升。所以懂调度机制的人从来不会只盯着 Slot 数量做规划而是把 Slot 数、CPU 核数、任务自身的 CPU 消耗放在一起算。2. Slot 不只是一个“坑位”它决定了整条流水线能铺多宽2.1 为什么 Flink 要用 Slot 作为资源分配单位Slot 是 TaskManager 对外提供的最小资源单位。一个 TaskManager 启动后会根据taskmanager.numberOfTaskSlots配置创建固定数量的 Slot并向 ResourceManager 注册这些 Slot 的信息。为什么用 Slot 而不是直接按线程数或者内存大小来做资源抽象原因有三第一统一调度接口。JobManager 不需要去了解每台机器的内存余量和 CPU 余量只需要问“有没有空闲 Slot”。有就把任务塞过去没有就让任务等待。第二静态规划简化了资源分配。TaskManager 启动时 Slot 数就固定了整个集群可用资源量是明确的。对于 Standalone 集群尤其如此管理员可以提前规划好集群最大能跑多少并行度。第三Slot 内置了内存划分的逻辑。每个 Slot 在 TaskManager 内存模型中都对应一部分资源份额尤其是网络缓冲区和托管内存。任务分配到 Slot 上就能获得稳定可期的资源边界。但注意Slot 并不限制一个 TaskManager 内总共能执行的线程数。Flink 的任务线程并不是按 Slot 严格绑定的Slot 只是一个逻辑资源分区。2.2 numberOfTaskSlots 配多少才合理这是配置 TaskManager 时最常见的问题。先说结论没有一个适合所有场景的固定值但有一条原则是普遍适用的——Slot 数最好不要超过机器的 CPU 核数。如果任务是 CPU 密集型比如大量规则计算、复杂窗口聚合那 Slot 数应当等于 CPU 核数甚至更保守一点设置为核心数减一留出系统开销和 GC 的余地。如果任务是 IO 密集型比如大量读写外部存储、频繁序列化和网络传输那么 Slot 数可以超过 CPU 核数因为任务线程大部分时间在等待 IO并不会造成过多 CPU 竞争。我自己的习惯是8 核机器如果作业以实时 ETL 为主Slot 设成 8如果作业有重聚合和复杂状态Slot 设成 4。宁可让 Slot 不够也不要让线程过度竞争。因为 Slot 不够最多是作业排队等待线程竞争导致的背压和延迟抖动反而更难排查。还有一个容易忽略的点一个 TaskManager 的 Slot 数量对单个任务可获得的内存也有影响。虽然内存不是按 Slot 平均切分的但 Slot 越多意味着同时挤在这个进程里的任务就越多JVM 堆和托管内存的压力就越大。这个我在第 5 章展开讲。2.3 Slot 共享机制是省资源的利器也可能是拖垮性能的陷阱Flink 默认开启了 Slot 共享。也就是说来自同一个作业、属于同一个 Slot 共享组的多个任务可以被塞进同一个 Slot 里执行。举例一个作业由 Source、FlatMap、Sink 三个算子链组成并行度都是 4。如果不启用 Slot 共享它需要 12 个 Slot。启用共享后它只需要 4 个 Slot每个 Slot 里跑一个 Source 子任务、一个 FlatMap 子任务、一个 Sink 子任务组成一条完整的处理流水线。所以计算作业所需 Slot 数有一个简单公式在默认共享情况下作业需要的 Slot 数约等于所有算子中最大的并行度而不是并行度的总和。Slot 共享的好处是显而易见的减少空闲浪费让每个 Slot 都尽量被填满。但它也有副作用它会让资源问题变得隐蔽。比如你的作业最大并行度是 8集群也有 8 个 Slot看起来资源正好但每个 Slot 里其实挤了 3 个线程。如果这 3 个线程都算 CPU 密集那实际上仍然是 24 个线程抢 8 个核的资源。如果你希望某些算子不跟其他算子共享 Slot可以用slotSharingGroup把它们隔离到不同的资源组。这在混布 CPU 密集任务和 IO 密集任务时非常有用后面第 6 章的案例里我会细说。3. Parallelism 的多层设置链算子优先级、提交参数与配置默认值3.1 四级优先级以及最容易被忽略的反直觉行为Parallelism 的设置有四个层级从高到低分别是算子级别dataStream.map(...).setParallelism(4)执行环境级别env.setParallelism(4)客户端提交级别flink run -p 4配置文件级别flink-conf.yaml中的parallelism.default优先级高并不意味着层级越高越好它只代表覆盖关系。一旦你在算子级别用setParallelism固定了并行度那么执行环境、命令行参数和配置文件都无法改变它。这里有一个反直觉的点需要特别提醒如果代码里写了env.setParallelism(8)然后你在提交作业时用flink run -p 16最终生效的并不是 16而是 8。因为执行环境级别高于客户端级别。我实际遇到过这种情况平台侧的同学为了保险统一在提交命令里加了-p 16但作业代码里显式调用了env.setParallelism(8)。结果作业一直在 8 并行度下运行吞吐上不去排障的时候绕了很大一圈才发现是优先级问题。所以建议是不要在代码里随意写env.setParallelism除非你有充分的理由。更好的做法是保持执行环境不设置并行度把并行度交给提交端的参数来控制这样不同场景可以灵活切换。3.2 并行度不是越大越好几个隐藏约束很多刚接触 Flink 的人会误以为并行度越大吞吐越高。实际上并行度存在多个隐藏约束。第一个约束是 Kafka 分区数。如果你从 Kafka 读数据Source 的并行度通常不应该超过主题的分区数。一个并行实例可以消费一个或多个分区但如果 Source 并行度大于分区数必然会有实例拿不到数据造成空闲线程。第二个约束是数据倾斜风险。并行度越大数据被切分的粒度越细keyBy 之后分区之间的数据量越容易不均。比如 10 个 key 的数据量差异巨大你就算并行度设到 100也大概率只有少部分 Task 在忙。第三个约束是网络和序列化开销。并行度增加意味着 Task 之间的数据交换通道呈平方级增长。16 个上游 Task 和 16 个下游 Task 之间会产生 256 条逻辑通道每条通道都要分配缓冲区序列化对象的开销也会放大。第四个约束是检查点的协调成本。并行度越高检查点需要对齐的分区数就越多Barrier 传递的路径也更复杂。并行度从 4 提到 8检查点时延可能不是翻倍而是增加好几倍。所以并行度的选择要基于目标吞吐、数据分布和计算复杂度综合推导而不是简单地“调大试试”。3.3 在 Web UI 上如何核对并行度和 Slot 分配配置完并行度之后一定要去 Flink Web UI 上核对实际执行情况。只看代码里的并行度是不够的因为算子链合并后UI 上显示的是合并后链的并行度而不是单个算子的并行度。打开作业的“任务”页面你会看到若干个节点每个节点代表一个被合并后的算子链节点上会标出它的并行度。点进节点详情可以看到每个子任务Subtask的索引、所在的 TaskManager 地址和 Slot 信息。如果几个不同算子的子任务在同一个 TaskManager 上且 Slot 编号相同说明它们发生了 Slot 共享。如果你发现某个 TaskManager 上挤了比其他节点多很多的任务那就要检查是不是并行度配置或共享组设置不当。另外在“Task Managers”页面可以看到每个 TaskManager 的 Slot 总数和已经使用的 Slot 数。提交新作业之前顺手看一眼这个页面能避免不少“作业一直 SCHEDULED 不动”的尴尬。4. 从吞吐量反推并行度一个实时数仓作业的资源规划全过程4.1 先做单实例吞吐测试别拍脑袋定并行度资源规划的第一步不是打开配置文件写数字而是做单实例吞吐测试。我接过一个实时指标计算作业需求是处理 Kafka 里的交易行为日志峰值 8 万条/秒每条消息约 200 字节计算逻辑包括 JSON 解析、规则过滤、按用户 ID 聚合、更新状态、输出到下游存储。我先用单并行度跑了一遍压测结果稳定在 2 万条/秒左右主要是因为聚合逻辑里涉及多字段的状态读写。那么目标 8 万条/秒就需要至少 4 个并行实例。实际工程上还要留出检查点对齐、GC 暂停、下游抖动等余量所以我会在理论值基础上乘 1.5 到 2 倍定为 8 个并行度。这个流程可以简化为一个公式并行度 目标吞吐 / 单实例吞吐 × 安全系数其中安全系数一般取 1.5 ~ 2。压测环境越接近生产系数就可以取得越小反之则往大了取。4.2 根据关键路径拆解各算子并行度确定了整体并行度之后还要分阶段细化。还是上面那个作业它的处理链路是Kafka Source → JSON 解析与过滤 → 用户维度聚合 → 结果写入下游Kafka 主题有 16 个分区那么 Source 并行度设为 16 是自然的选择否则会出现部分分区长时间不被消费的情况。解析和过滤阶段是纯 CPU 密集逻辑单实例压测是 2 万条/秒所以 8 个并行度就能满足 8 万条/秒的峰值并且留有足够余量。用户维度聚合阶段涉及状态读写单实例吞吐进一步下降到 1.2 万条/秒需要约 7 个并行度才能覆盖 8 万条/秒取整到 8。结果写入下游的 Sink受目标库写入能力限制并行度设为 4并且通过批量写入来控制压力。所以这个作业最终各阶段的并行度分别是 16、8、8、4。由于默认 Slot 共享生效整个作业需要的总 Slot 数不等于 1688436而是最大并行度 16。但这里我会特意不靠 Slot 共享把一切都塞进 16 个 Slot因为聚合阶段和 Source 的负载模式差异很大混在一起容易互相干扰。我会把聚合阶段单独放到一个 Slot 共享组里给它独立的 Slot 配额。4.3 拿到并行度后换算所需 Slots 和 TaskManager 数量确认并行度之后再往下推资源。我当时的集群是 4 台 8 核机器。如果每台机器跑 1 个 TaskManager、每个 TaskManager 设 4 个 Slot那么总 Slot 数是 4×416。作业使用 16 个 Source 并行度的话刚好能把集群的 Slot 全部填满。但这样太满了没给波动留空间。所以我最终每台机器只设 3 个 Slot总 Slot 数是 12作业把聚合阶段单独分成共享组组内并行度 8。这样 12 个 Slot 中聚合组占 8 个剩下 4 个给 Source、解析、Sink 用仍然满足要求。CPU 核数 8、Slot 数 3每个 Slot 平均能摊到约 2.7 个核线程竞争压力也能接受。这个案例的启发是资源规划不能只算“够不够”还要算“留了多少余地”。集群资源用满不等于运行稳定尤其在峰值流量和检查点叠加的时候资源要有冗余。5. TaskManager 内存与 Slot 的耦合点多配 Slot 不等于多分内存5.1 TaskManager 内存模型快速认识Flink 从 1.10 开始使用统一内存配置模型一个 TaskManager 进程的总内存由taskmanager.memory.process.size指定内部再划分成多个区域。从 JVM 角度看内存分成两部分一部分是 JVM 堆内存包括框架堆内存和任务堆内存。任务堆内存是用户代码和普通对象的主要活动区域。另一部分是堆外内存包括框架堆外内存、任务堆外内存、托管内存、网络内存和 JVM Overhead。其中托管内存最有 Flink 特色它被 Flink 自行管理主要分配给排序、哈希表和 RocksDB 状态后端使用。网络内存则用于任务之间的数据交换缓冲。配置项方面常用的有taskmanager.memory.process.size进程总内存taskmanager.memory.managed.size托管内存大小也可以是托管内存占比taskmanager.memory.network.min/taskmanager.memory.network.max网络内存范围taskmanager.memory.jvm-overheadJVM 运行额外开销一个常见的启动参数示例是-Dtaskmanager.memory.process.size4096m让每个 TaskManager 进程拥有 4GB 总内存。内部各区域占比如果没显式指定会按默认值自动分配。5.2 Slot 数量变化对单个任务可用内存的真实影响很多人会以为4GB 内存、4 个 Slot那么每个 Slot 就有 1GB 内存可用。严格来说这是不对的。堆内存不是按 Slot 平均切开的。任务堆内存是一个共享池所有运行在同一个 TaskManager 里的任务都从同一个堆里申请对象。托管内存同样是一个进程级共享池RocksDB 多个列族、多个任务共用这份内存预算。所以当你把 Slot 数从 4 增加到 8 时单个任务获得的内存并不会自动翻倍或减半但并发任务数会改变内存竞争的激烈程度。举个例子我的一个作业使用 RocksDB 状态后端TaskManager 总内存 8GB托管内存默认占 0.4也就是约 3.2GB。并行度 4 时4 个聚合 Task 共享 3.2GB 托管内存每个任务实际能用的上限是充足的。后来我把并行度调到 8还是同一个 TaskManager相同的内存配置8 个任务共用 3.2GB单个任务的内存可用性一下变得紧张State 频繁在磁盘和内存之间交换性能明显下降。遇到这种情况正确的做法不是继续调大并行度而是要么增加 TaskManager 总内存要么减少每个 TaskManager 的 Slot 数让任务分散到更多进程中去。5.3 网络缓冲、反压与并行度之间的联动内存模型里还有一个容易被忽视的角色网络内存。两个 Task 之间传输数据不是直接共享变量而是通过有界缓冲区来传递。上游 Task 把序列化后的数据写入网络缓冲区下游 Task 从缓冲区读取。如果下游处理不过来缓冲区被写满上游就自然停下来等待这就是反压。并行度越高Task 之间建立的逻辑通道越多需要的网络缓冲区总量也越大。比如上游并行度 16、下游并行度 16那就可能产生 256 条逻辑通道。虽然 Flink 会通过共享缓冲区机制优化但整体压力仍然是会上升的。如果你看到作业里频繁出现Channel closed、Buffer pool exhausted、Task is backpressured排除了下游处理逻辑的问题之后要想到可能不是 CPU 不够而是网络内存配置偏低或并行度太高导致的缓冲区碎片化。可以试着把taskmanager.memory.network.min和max适当调大或者通过-Dtaskmanager.memory.network.max512m这类参数调优。但更根本的方法还是检查并行度设置是否合理减少不必要的跨 Task 数据通道。6. 踩过的坑与提交前检查清单6.1 三个典型的排障复盘第一个坑作业一直处于 SCHEDULED 状态等多久都不跑。当时的表现是TaskManager 有几个空闲但新作业就是无法启动。查到最后发现全局的 Slot 总数小于作业的最大并行度。更具体地说我当时的作业最大并行度是 24但集群里只有 20 个可用 Slot而且代码里也没有设置 Slot 共享组调度器只能排队等资源。这类问题在看日志时会有大量类似“Not enough TMs available”的提示。但实际上集群经常不是真的没有机器而是机器能提供的 Slot 数不够。解决方法是调整并行度或者给集群扩容实在不行就把多余的重算子拆成独立作业分散调度。第二个坑并行度从 12 调到 48 之后吞吐没涨反而下降。这是一个状态聚合作业机器 8 核每个 TaskManager 8 个 Slot。并行度 48 意味着 6 个 TaskManager 上满负载跑 48 个聚合任务。结果每个任务处理的数据被切得更细不同 Subtask 之间需要通过网络做更多的 shuffle序列化和反序列化开销暴增。加上检查点需要对 48 个任务做对齐状态一致性的同步成本也更高。最终总吞吐还不如 24 并行度的时候。之后我把并行度降回 32并且调整了 keyBy 的逻辑让数据分布更均匀。吞吐才重新回到预期线。这个案例给到我的教训是并行度最好是“算出来的”不是“试出来的”。试的过程中要多观察每单位资源的产出效率。第三个坑Slot 共享组失效导致某个 TaskManager 超负荷。当时我在一个作业里同时跑了 CPU 密集的复杂规则计算和大量读写的 Sink。默认情况下它们共享 Slot结果规则计算 Task 和 Sink Task 在同一个 Slot 里互相抢资源背压指标时好时坏。后来我用slotSharingGroup把规则计算单独隔离成一个组给它分配独立的 SlotSink 放到另一个组又调节了各组的并行度。整个作业的延迟曲线立刻平稳下来。这个案例说明Slot 共享不是不能拆而是要有意识地拆在资源边界上。6.2 提交任务之前我必做的六步检查我现在每次提交 Flink 作业前都会按固定流程过一遍避免低级问题影响线上稳定性。第一写清楚每个算子的并行度及其依据。Source 是不是等于分区数聚合算子有没有压测过单吞吐Sink 并行度是否匹配下游写入能力。第二计算作业需要的 Slot 数。默认共享下取最大并行度显式划分了共享组的情况下需要把不同组需要的 Slot 数相加。第三核对集群当前可用 Slot 数。这一步要看实时指标而不是只查总配置。因为可能还有别的作业正在占用资源新作业的调度一定要在当前空闲 Slot 基础上判断。第四把 Slot 数与 CPU 核数做对比。确认每个 Slot 平均能分配到的核数不要低于 1。如果 Slot 数超过核数要评估任务里是否大部分是 IO 等待否则就属于过度订阅。第五检查内存配置。确认 TaskManager 总内存能否支撑当前 Slot 数下的所有任务尤其是状态后端的托管内存不要让多个任务共享一份并不宽裕的内存池。第六确认好提交参数与代码里并行度设置的优先级。检查代码里有没有setParallelism这种隐式覆盖避免命令行-p传了也白传。6.3 几个简单换算公式贴到笔记里随时用最后分享几个我平时一直在用的公式它们不复杂但能让你在排查资源问题时省下很多时间。总 Slot 数 所有 TaskManager 的 Slot 数之和。如果所有 TaskManager 规格相同就是 TaskManager 数乘以单机 Slot 数。作业最少所需 Slot 数在不拆共享组时约等于作业中最大并行度。如果拆了 Slot 共享组则需要把每个共享组内所需 Slot 数分别算出来后相加。CPU 过度订阅倍数 总 Slot 数 / 总 CPU 核数。这个值超过 1 时说明存在线程竞争风险需要确认任务是不是 IO 密集型。并行度推算 目标吞吐 / 单实例基准吞吐 × 安全系数安全系数建议 1.5 ~ 2。这里想再强调一次公式只是入口关键还是对作业自身的计算特征有感知。同样的吞吐目标一个纯过滤清洗的作业和一个带大状态聚合的作业所求出的并行度可能差好几倍。资源规划如果真的想做好唯一的笨办法就是把每个算子的计算逻辑都摸透然后交给压测数据来验证。这是我做了几年实时计算之后最想分享的一条经验。