复杂聚合产生大量回撤流（Retract），导致下游 Sink 写入瓶颈，如何解决？

在 Flink SQL 中，复杂的聚合和关联（如 GROUP BY、JOIN 等）会导致 Flink 产生大量的回撤流（Retract Stream，即 -U / UPDATE_BEFORE 和 +U / UPDATE_AFTER 报文）。当这些高频的回撤数据写入下游 Sink 时，目标存储（如 MySQL、Elasticsearch、HBase 等）需要频繁执行删除/更新与插入操作，从而产生极大的写入瓶颈。

针对这一问题，我们可以从 Flink 引擎层优化、Sink 端与下游优化、SQL 开发及架构优化 几个维度来系统性地解决。

一、 Flink SQL 引擎层优化（源头减少回撤消息）

通过调整 Flink 优化器的配置，让 Flink 在内存中合并和消化一部分中间更新状态，从而大幅减少发送给下游 Sink 的回撤消息数量。

1. 开启 MiniBatch 攒批机制

默认情况下，Flink SQL 的聚合算子是每来一条数据就处理并向下游发送一条数据。开启 MiniBatch 后，Flink 会在内存中缓存一定时间或一定数量的数据，合并后再进行状态更新和下发，这能极大减少回撤消息的发生频率。

sql

-- 开启 MiniBatch
SET 'table.exec.mini-batch.enabled' = 'true';
-- 允许的延迟时间，例如 5 秒（根据业务对实时性的要求调整）
SET 'table.exec.mini-batch.allow-latency' = '5s';
-- 每个批次的最大缓存条数
SET 'table.exec.mini-batch.size' = '10000';

2. 启用 Local-Global 两阶段聚合

类似于 MapReduce 中的 Combine 阶段，Local-Global 优化会在 Local 算子对本地的微批（MiniBatch）数据进行预聚合，然后再将结果发送到 Global 算子进行全局聚合。这能大幅度降低网络 Shuffle 带来的开销，并显著减少向下游发送的更新报文。
注：该特性依赖于 MiniBatch 的开启。

sql

-- 开启 Local-Global 优化（默认通常为 AUTO，显式开启可确保生效）
SET 'table.optimizer.agg-phase-strategy' = 'TWO_PHASE';

3. 拆分 Distinct 聚合（Split Distinct）

如果你的 SQL 中包含高基数的 COUNT(DISTINCT) 聚合，会导致严重的倾斜与高频回撤。开启此配置后，Flink 会自动将 Distinct 拆分为两层聚合（第一层通过 Hash 散列去重，第二层再做全局聚合），从而分散压力并平滑回撤流。

sql

-- 开启 Distinct 自动拆分
SET 'table.optimizer.distinct-agg.split.enabled' = 'true';

二、 Sink 与下游存储优化（降低写入压力）

1. 调整 `upsert-materialize` 配置

Flink 为了保证下游多流关联或乱序情况下的结果确定性，默认会在 Sink 之前自动加上一个 SinkUpsertMaterializer 算子（其配置默认为 AUTO）。该算子会在 Flink 内部维护一份状态，用来确保下发给 Sink 的主键数据不发生乱序，但这可能会产生额外的回撤消息。

如果你的数据源本身能够保证顺序，或者下游 Sink 本身具有幂等性且对短暂的乱序更新不敏感（例如简单的 KV 覆盖写），可以尝试将该物化配置关闭，以减少内部状态开销和多余的回撤：
sql
```
SET 'table.exec.sink.upsert-materialize' = 'NONE';
```

2. 使用 Upsert Sink 替代 Retract Sink

Retract Sink 会将一次更新拆分为 -U（删除旧数据）和 +U（写入新数据）发送给目标端。
Upsert Sink（如带有主键的 JDBC、Elasticsearch、MongoDB 连接器等）能够感知 Upsert 语义。如果在 DDL 中合理定义了主键（Primary Key），它们在很多情况下会直接合并 -U 与 +U，直接向目标库发起 UPSERT（如 MySQL 的 ON DUPLICATE KEY UPDATE），避免了先 delete 后 insert 的双倍 IO 压力。

3. 调大 Sink 端的攒批与刷写参数

在下游 Sink 的 DDL WITH 参数中，适当调大批次写入的大小和刷写间隔。例如，在 JDBC 或 Elasticsearch 连接器中：

sql

CREATE TABLE sink_table (
  id INT PRIMARY KEY NOT ENFORCED,
  val VARCHAR
) WITH (
  'connector' = 'jdbc',
  ...
  'sink.buffer-flush.max-rows' = '5000',     -- 攒满多少条再写入
  'sink.buffer-flush.interval' = '2s'        -- 或者每隔 2 秒刷写一次
);

4. 引入异步合并机制（如 ClickHouse ReplacingMergeTree）

如果下游是 ClickHouse，尽量避免让 Flink 执行高频的回撤更新。

可以将 Flink 任务改造为 Append-only 写入（只发送累加流）。
下游 ClickHouse 采用 ReplacingMergeTree 或 CollapsingMergeTree 引擎，由 ClickHouse 在后台异步合并相同主键的数据，从而彻底消除 Flink 端的回撤开销。

三、 SQL 开发与架构优化（从根本上规避回撤）

1. 尽量用窗口聚合（Window Aggregation）替代无界分组聚合（Group By）

GROUP BY key 会在状态中无限期保留 key，并随数据的到来不停地触发回撤和更新。
如果业务场景允许，建议使用 滚动窗口（Tumble）、滑动窗口（Hop） 或 累积窗口（Cumulate） 进行聚合。窗口聚合通常只在窗口结束（或触发 Watermark 推进）时输出一次结果，这属于 Append-only 流，基本不会产生回撤流。

2. 避免大状态的 Regular Join

常规的双流 JOIN（如 A LEFT JOIN B ON A.id = B.id）会在 Flink 状态中保存双方的所有数据，任何一侧数据的变化都会引起级联回撤，从而导致下游极大的吞吐瓶颈。

替代方案 1：如果右表是配置表或维度表，使用 Lookup Join（维表关联） 代替双流 Join。
替代方案 2：如果是有时效性的关联，使用 Interval Join（区间关联），这能让 Flink 自动清理过期状态，减少因为历史数据变更产生的回撤。

3. 设置合理的 State TTL

无界聚合会导致 Flink 状态无限增长，进而导致内存垃圾回收（GC）变慢、吞吐量下降，间接放大了下游 Sink 的写入延迟。务必为状态设置合理的生存时间（TTL），让过期的 Key 自动失效，保持 Flink 内部状态的健康。

sql

-- 状态保留 24 小时（根据业务实际需要设定，如 1d、12h 等）
SET 'table.exec.state.ttl' = '24h';

总结优化路线推荐

在实际调优时，建议遵循以下顺序进行排查与调整：

第一步： 开启 MiniBatch 并配合 Local-Global（这两个配置对业务入侵最小，通常能直接降低 50% 以上的回撤流量）。
第二步： 确认下游 Sink 的 DDL 是否定义了主键，并调大其缓冲刷写参数（buffer-flush 等）以提高单次写入的吞吐。
第三步： 排查执行计划（使用 EXPLAIN），如果存在不必要的 SinkUpsertMaterializer，且下游能容忍轻微乱序覆盖，将其设为 NONE。
第四步： 如果依然无法承载，需从业务角度审视，是否可以将 unbounded 的 GROUP BY 优化为 Window 聚合，或者将双流关联优化为维表关联/区间关联。

右滑查看面试常问

播面