Flink 自适应批执行（Adaptive Batch Execution）让 Batch 作业“边跑边优化”

1. 自适应批执行解决的核心痛点

传统静态计划的问题不在于优化器不聪明，而在于“信息不够”：

• 输入数据统计经常缺失或不准
• 中间数据量和分布要等跑起来才知道
• Join 的两侧大小变化大，今天广播是神优化，明天可能直接 OOM
• 并发度每天都要重新估，尤其是“每天数据量波动”的离线链路

自适应批执行的思路是：别强行在开跑前把所有决策做完，让作业跑起来拿到真实数据特征，再做决定。

2. AdaptiveBatchScheduler 能做哪些“运行时优化”

当前支持的策略包括：

1. 自动决定算子并发度（Auto Parallelism）
2. 自动做数据分布负载均衡（Automatic Load Balancing）
3. 自适应广播 Join（Adaptive Broadcast Join）
4. 自适应倾斜 Join 优化（Adaptive Skewed Join Optimization）

下面逐个讲清楚“它做了什么、怎么用、什么时候要注意”。

3. 自动决定算子并发度：把最耗人的并发调优交给调度器

3.1 它怎么决定并发度

如果某个算子没有显式设置 parallelism，调度器会根据它消费的数据集大小推导并发度。收益很直接：

• 你不用每天盯着并发调参
• 数据量每天波动时，并发也能跟着自适应
• SQL Batch 作业里，不同算子可以拿到不同并发度（更贴合真实数据体量）

3.2 使用要点：想让它管，就别手动管

关键原则只有一个：不要对你希望自适应的算子调用setParallelism()。因为它只会对“未指定并发度”的算子做推导。

3.3 关键配置项（建议你至少看一遍）

# 总开关（默认开启）execution.batch.adaptive.auto-parallelism.enabled:true# 自适应并发下限/上限execution.batch.adaptive.auto-parallelism.min-parallelism:1execution.batch.adaptive.auto-parallelism.max-parallelism:256# 期望每个 Task 平均处理的数据量（影响推导结果）execution.batch.adaptive.auto-parallelism.avg-data-volume-per-task:256mb# Source 默认并发（或 Source 自适应并发的上限）execution.batch.adaptive.auto-parallelism.default-source-parallelism:64

关于max-parallelism的直觉很重要：不是越大越好。并发上限过高会带来大量 subpartition，可能拖慢 hash shuffle 与网络传输（小包变多、开销变大）。更合理的做法是：把它设置成你“最坏情况下”真正需要的上限。

3.4 Source 的动态并发推导（高级用法）

新 Source 可以实现DynamicParallelismInference接口，让 Source 在调度前根据上下文推导并发：

publicinterfaceDynamicParallelismInference{intinferParallelism(Contextcontext);}

Context 会给你：

• 允许的并发上限
• 期望每个 task 处理的平均数据量
• 动态过滤信息（dynamic filtering），帮助你更精准推导

注意：这个推导会在调度源算子前调用，实现里要避免耗时操作，否则会拖慢调度。

如果 Source 没实现该接口，则使用execution.batch.adaptive.auto-parallelism.default-source-parallelism作为 source 并发（前提是 source 本身没被手动 setParallelism）。

4. 自动负载均衡：让下游“吃得更均匀”

调度器会尽量把数据均匀分到下游 subtasks，目标是让每个下游 subtask 消费的数据量差不多，减少“有的忙死、有的闲死”的情况。它适用于多种连接边：

• point-wise（例如 Rescale）
• all-to-all（Hash / Rebalance / Custom）

重要限制：目前它只支持“启用了自动并发推导”的算子。也就是说：

• 想吃到负载均衡红利 → 先开 auto parallelism → 别手动设 parallelism

它也解决不了“单 key 超级热点”的问题，因为为了正确性，单 key 的数据不能随便拆给不同 subtask。但这类问题在某些 Join 场景会被“自适应倾斜 Join 优化”部分缓解。

5. 自适应 Broadcast Join：别再靠静态统计“赌”广播

5.1 为什么需要它

广播 Join 很香：小表广播到每个节点，Join 在内存里做，省掉大表 shuffle/sort，速度飞起。

但静态优化很容易误判：

• 生产里源表统计不准
• 更糟的是 Join 输入可能来自中间结果，运行前根本没法评估大小
• 一旦把“大表误判成小表”走广播，可能直接 OOM（构建 hash 表爆内存），任务重启，代价巨大

自适应 Broadcast Join 的价值在于：运行时看真实输入大小，再决定要不要把 Join 转成广播。

5.2 哪些 Join 类型允许广播（语义正确性约束）

• Inner：左右都可广播
• LeftOuter：只能广播右侧
• RightOuter：只能广播左侧
• FullOuter：两侧都不允许
• Semi / Anti：只能广播右侧

5.3 配置与策略

调度器默认同时启用“编译期静态广播”和“运行期自适应广播”。你可以控制只在运行时生效：

table.optimizer.adaptive-broadcast-join.strategy:RUNTIME_ONLY

阈值配置（决定多大算“小表”）：

table.optimizer.join.broadcast-threshold:64mb

TaskManager 内存大可以适当提高阈值；内存紧张就降低，避免运行时广播把内存顶爆。

5.4 限制

• MultiInput 算子内部的 Join 不支持
• 不能与 Batch Job Recovery Progress 同时启用（启用恢复进度后，自适应广播不生效）

6. 自适应倾斜 Join 优化：专治 Join 尾延迟

Join 最怕数据倾斜：某些 key 极高频，导致对应 Join task 处理量远超其他 task，出现明显尾延迟，拖慢整个 stage 完成。

由于 Join 的关联性，简单“负载均衡”无法把同一个 keyGroup 拆到不同 task（否则结果不正确）。自适应倾斜 Join 的思路是：根据运行时统计，把倾斜且可拆分的分区动态切分，缓解尾延迟。

6.1 哪些 Join 类型支持动态拆分

• Inner：左右都可拆分
• LeftOuter：只能拆分左侧
• RightOuter：只能拆分右侧
• FullOuter：都不支持
• Semi / Anti：只能拆分左侧

6.2 策略控制

table.optimizer.skewed-join-optimization.strategy:auto

可选值：

• none：关闭
• auto：尽量启用，但如果需要额外 shuffle 才能保证正确性，则为了避免开销不会生效
• forced：即使引入额外 shuffle 也强制生效

阈值与因子（调到适合你的作业特征）：

table.optimizer.skewed-join-optimization.skewed-threshold:256mbtable.optimizer.skewed-join-optimization.skewed-factor:4.0

直觉解释：

• threshold：大到什么程度算“触发倾斜优化”
• factor：把“最大/中位数”的比例压到多少以下算“够均衡”

6.3 限制

• 目前要求启用“自动并发推导”，因为它可能影响 Join 算子并发
• MultiInput 内的 Join 不支持
• 不能与 Batch Job Recovery Progress 同时启用

7. 性能调优建议：让自适应更稳、更不容易炸网内存

官方给了两个非常实用的建议：

1. 推荐使用 Sort Shuffle，并设置：

taskmanager.network.memory.buffers-per-channel:0

这样能把网络内存需求与并发解耦，大规模作业更不容易报 “Insufficient number of network buffers”。

2. execution.batch.adaptive.auto-parallelism.max-parallelism建议设成你预期的“最坏情况上限”，不要无限大
   上限过大可能导致 subpartition 数过多，影响 hash shuffle 性能与网络传输（小包变多、开销变大）。

8. 使用边界：什么情况下它根本不会生效

• 必须使用 AdaptiveBatchScheduler（它是默认 batch scheduler，除非你手动改成别的，例如jobmanager.scheduler: default）
• 只支持 BLOCKING / HYBRID 作业（ALL_EXCHANGES_BLOCKING / ALL_EXCHANGES_HYBRID_FULL / ALL_EXCHANGES_HYBRID_SELECTIVE）
• 不支持 FileInputFormat（例如readFile(...)/createInput(FileInputFormat, ...)），要用新 Source（FileSystem DataStream Connector / FileSystem SQL Connector）
• Web UI 的 broadcast 发送/接收指标可能不一致（自动并发推导场景下会让人困惑）

9. 一套落地建议：从“可控收益”开始启用

如果你要在生产逐步引入，建议按这个顺序：

1. 先只启用自动并发推导（少改代码收益大）

• 移除或避免对算子setParallelism()
• 配好 min/max/avg-data-volume-per-task

2. 观察是否出现网络 buffer 压力或 subpartition 激增

• 适当收紧 max-parallelism
• 考虑 Sort Shuffle + buffers-per-channel=0

3. 再逐步启用自适应 Broadcast Join（收益很大，但要管住阈值）

• 内存紧张先把 broadcast-threshold 调小

4. 最后再开倾斜 Join 优化（对“尾延迟拖全局”的作业非常有价值）

• auto 起步，必要时 forced，但要评估额外 shuffle 代价