【Flink 30天】Day22-23 FlinkSQL 性能优化：Mini-Batch + 两阶段聚合 + TOP-N + 完整配置

Day 22-23：FlinkSQL 性能优化

1. Mini-Batch（微批优化）

问题：FlinkSQL 默认每条数据到来就触发一次状态读写，高 QPS 下频繁 IO 性能差。

Mini-Batch：积累一批数据，一次性处理，减少状态访问次数。

// 开启 Mini-Batch
TableConfig config = tEnv.getConfig();
config.set("table.exec.mini-batch.enabled", "true");
config.set("table.exec.mini-batch.allow-latency", "5 s"); // 最多等 5 秒
config.set("table.exec.mini-batch.size", "5000"); // 或凑够 5000 条

效果：对聚合操作有 5~10 倍提升，延迟从毫秒级变为秒级（trade-off）。

2. 两阶段聚合（Local-Global Aggregation）

问题：数据倾斜时，某个 key 的所有数据都发到一个 SubTask 处理，单点瓶颈。

两阶段聚合：

• Local 阶段：先在每个 SubTask 本地做部分聚合（打散热 key）
• Global 阶段：再汇总 Local 的结果

// 开启两阶段聚合（需要先开启 Mini-Batch）
config.set("table.optimizer.agg-phase-strategy", "TWO_PHASE");

原来（一阶段）：
数据 → Exchange(hash by userId) → GroupAggregate
热 key "user_001" 全部压到 SubTask 3

两阶段：
数据 → LocalGroupAggregate → Exchange(hash by userId) → GlobalGroupAggregate
热 key "user_001" 先在各 SubTask 局部合并，再汇总 → 均匀分散了压力

3. 去重优化

场景：对一个 Stream 去重，只保留每个 key 的最新一条（或第一条）。

低效写法（产生大量状态）

— 全量 ROW_NUMBER，状态保留所有历史数据
SELECT userId, amount, ts FROM (
SELECT *, ROW_NUMBER() OVER (PARTITION BY userId ORDER BY ts DESC) AS rn
FROM events
) WHERE rn = 1;

优化写法

// 方案1：开启 Deduplication 优化（Flink 自动识别并优化上面的 SQL）
config.set("table.optimizer.distinct-agg.split.enabled", "true");

// 方案2：如果只需要第一条（追加）
SELECT userId, FIRST_VALUE(amount), MIN(ts) FROM events GROUP BY userId;

真正高效的方案：状态 TTL + 幂等写入

// 给 TableEnv 设置全局 State TTL
config.set("table.exec.state.ttl", "86400 s"); // 1天

4. TOP-N 优化

场景：每分钟统计各品类销售额 TOP 10

— TVF 窗口 + TOP-N（推荐写法，有状态优化）
SELECT category, product, amount, window_start, window_end
FROM (
SELECT *,
ROW_NUMBER() OVER (
PARTITION BY window_start, window_end, category
ORDER BY amount DESC
) AS rn
FROM (
SELECT category, product,
SUM(amount) AS amount,
window_start, window_end
FROM TABLE(TUMBLE(TABLE sales, DESCRIPTOR(ts), INTERVAL '1' MINUTES))
GROUP BY window_start, window_end, category, product
)
)
WHERE rn <= 10;

注意：

• 外层 ROW_NUMBER 必须在 window_start, window_end 上 PARTITION，否则会产生无界状态
• 结合窗口使用，窗口关闭后状态自动清除

5. 内置优化器配置

// 谓词下推（自动，无需配置）
// Flink 会把 WHERE 条件尽量推到 Source 端过滤，减少传输数据量

// 投影下推（自动）
// 只读取 SQL 中用到的列

// 开启多阶段聚合
config.set("table.optimizer.agg-phase-strategy", "AUTO"); // 自动选择

// 开启动态过滤（Join 时用小表过滤大表）
config.set("table.optimizer.runtime-filter.enabled", "true");

6. 完整配置参考

TableConfig config = tEnv.getConfig();

// Mini-Batch
config.set("table.exec.mini-batch.enabled", "true");
config.set("table.exec.mini-batch.allow-latency", "5 s");
config.set("table.exec.mini-batch.size", "5000");

// 两阶段聚合
config.set("table.optimizer.agg-phase-strategy", "TWO_PHASE");

// 全局 State TTL（防止无界状态增长）
config.set("table.exec.state.ttl", "86400 s");

// 去重优化
config.set("table.optimizer.distinct-agg.split.enabled", "true");

小结