1. 自适应批执行解决的核心痛点传统静态计划的问题不在于优化器不聪明而在于“信息不够”输入数据统计经常缺失或不准中间数据量和分布要等跑起来才知道Join 的两侧大小变化大今天广播是神优化明天可能直接 OOM并发度每天都要重新估尤其是“每天数据量波动”的离线链路自适应批执行的思路是别强行在开跑前把所有决策做完让作业跑起来拿到真实数据特征再做决定。2. AdaptiveBatchScheduler 能做哪些“运行时优化”当前支持的策略包括自动决定算子并发度Auto Parallelism自动做数据分布负载均衡Automatic Load Balancing自适应广播 JoinAdaptive Broadcast Join自适应倾斜 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例如 Rescaleall-to-allHash / 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:64mbTaskManager 内存大可以适当提高阈值内存紧张就降低避免运行时广播把内存顶爆。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. 性能调优建议让自适应更稳、更不容易炸网内存官方给了两个非常实用的建议推荐使用 Sort Shuffle并设置taskmanager.network.memory.buffers-per-channel:0这样能把网络内存需求与并发解耦大规模作业更不容易报 “Insufficient number of network buffers”。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, ...)要用新 SourceFileSystem DataStream Connector / FileSystem SQL ConnectorWeb UI 的 broadcast 发送/接收指标可能不一致自动并发推导场景下会让人困惑9. 一套落地建议从“可控收益”开始启用如果你要在生产逐步引入建议按这个顺序先只启用自动并发推导少改代码收益大移除或避免对算子setParallelism()配好 min/max/avg-data-volume-per-task观察是否出现网络 buffer 压力或 subpartition 激增适当收紧 max-parallelism考虑 Sort Shuffle buffers-per-channel0再逐步启用自适应 Broadcast Join收益很大但要管住阈值内存紧张先把 broadcast-threshold 调小最后再开倾斜 Join 优化对“尾延迟拖全局”的作业非常有价值auto 起步必要时 forced但要评估额外 shuffle 代价