20250124 Flink 增量聚合 vs 全量聚合

1. 增量聚合 vs 全量聚合

(1) 增量聚合（ReduceFunction / AggregateFunction）

工作方式：
- 逐步计算：每一条数据到达窗口时，立即与当前聚合结果结合，生成新的中间结果。
- 仅保存中间状态：内存中只保留当前的聚合值（如累加和、最大值等），不保存原始数据。
- 触发窗口计算时：直接输出最终的聚合结果，无需遍历所有数据。

示例：计算窗口内数字的和

DataStream<Integer> numbers = ...;
numbers.keyBy(...).window(...).aggregate(new AggregateFunction<Integer, Long, Long>() {// 初始值：累加器初始化为 0public Long createAccumulator() { return 0L; }// 增量操作：每来一个数，累加到当前和public Long add(Integer value, Long accumulator) {return accumulator + value;}// 输出最终结果public Long getResult(Long accumulator) { return accumulator; }// 合并多个累加器（仅会话窗口需要）public Long merge(Long a, Long b) { return a + b; }});

内存占用：无论窗口内有多少数据，状态中始终只保存一个 Long 类型的累加值（极低开销）。

(2) 全量聚合（ProcessWindowFunction）

工作方式：
- 缓存所有数据：窗口触发前，所有原始数据被保存到状态中。
- 触发窗口计算时：遍历所有数据并一次性处理（例如排序、求中位数等复杂操作）。
- 内存开销：状态大小与窗口内数据量成正比（可能引发 OOM）。

示例：获取窗口内所有数据

DataStream<Integer> numbers = ...;
numbers.keyBy(...).window(...).process(new ProcessWindowFunction<Integer, String, Key, TimeWindow>() {public void process(Key key, Context ctx, Iterable<Integer> elements, Collector<String> out) {List<Integer> list = new ArrayList<>();for (Integer num : elements) {list.add(num); // 需要遍历所有数据}out.collect("窗口数据: " + list);}});

内存占用：若窗口内有 100 万条数据，状态中将保存 100 万个 Integer 对象。

2. 为什么增量聚合更高效？

(1) 内存开销低

增量聚合：仅保存聚合中间结果（如 sum=100, max=50）。
全量聚合：保存所有原始数据（如 [10, 20, 30, ...]）。

(2) 计算效率高

增量聚合：每条数据触发一次简单计算（如加法）。
全量聚合：窗口触发时需遍历所有数据，时间复杂度为 O(n)。

(3) 避免数据堆积

增量聚合：实时输出中间结果，适合高吞吐场景。
全量聚合：窗口触发时可能因数据量大导致延迟。

3. 适用场景对比

场景	增量聚合	全量聚合
简单聚合（求和、计数、最大/最小）	✅ 优先使用（高效、低内存）	❌ 不必要
复杂计算（中位数、方差）	❌ 无法直接实现（需自定义复杂逻辑）	✅ 必须使用（需遍历所有数据）
需要访问窗口元信息（起止时间）	❌ 无法直接获取（需结合 `ProcessWindowFunction`）	✅ 直接通过 `Context` 获取

4. 综合使用：增量聚合 + ProcessWindowFunction

如果需要 同时实现高效聚合和访问窗口元信息，可将两者结合：

DataStream<Integer> numbers = ...;
numbers.keyBy(...).window(...).aggregate(new AggregateFunction<Integer, Long, Long>() { /* 增量求和 */ },new ProcessWindowFunction<Long, Result, Key, TimeWindow>() { // 补充元信息public void process(Key key, Context ctx, Iterable<Long> iterable, Collector<Result> out) {Long sum = iterable.iterator().next(); // 获取增量聚合结果out.collect(new Result(key, sum, ctx.window().getStart(), ctx.window().getEnd()));}});