运营要一张看板：每个城市近 30 天的成交额、最大一笔订单、平均订单金额和下单人数。四个指标写成 SQL 只是一行 GROUP BY，上线后却各慢各的：成交额每天把 30 个分区重扫一遍，其中 29 个昨天刚扫过；下单人数的 COUNT(DISTINCT user_id) 把一个任务的内存撑爆；北京的订单占了一半，整个作业都在等北京那一个任务。

站里有三篇文章分别讲了这三件事：消灭暴力扫描的递推、大数据去重统计的 HyperLogLog、大数据性能调优的两阶段聚合。分开看是三套技巧。这篇回答一个问题：它们是不是同一件事？

我的答案是：是。几乎所有聚合上的优化，都在利用聚合函数的两个性质：能不能合并，能不能撤销。一个指标有哪个性质，就能用哪一类优化；没有，就换一种数据结构让它有。

一张订单表，四个指标

全文只用一张按天分区的订单表 orders(dt, city, user_id, amount)，算每个城市近 30 天的四个指标：

下面先用两个问题给这四个指标分类，后面每一种优化，都只是在用这两个问题的答案。

两个问题：能不能合并，能不能撤销

能合并：拆开各算，再合起来

把 30 天的订单按天切开，每天算一个结果，再把 30 个结果合起来，等不等于一次算全部？

• 成交额：30 个日成交额相加。能。

• 最大单笔：30 个日最大值再取最大。能。

• 平均订单金额：30 个日平均值再平均，不对。第一天 1 单 100 元，第二天 3 单各 20 元：两个日平均是 100 和 20，再平均得 60；真实的平均是 (100 + 60) / 4 = 40。订单多的那天被少算了。但如果每天存的不是平均值，而是金额之和与订单数两个数，合并时各自相加再相除，就对了。

• 下单人数：30 个日人数相加，会把连续几天都下单的人算好几遍。要合并，每天必须保存「到底是哪些人」，中间状态和明细几乎一样大。

Jim Gray 等人在 1997 年的 Data Cube 论文里把聚合函数分成三类，正好对应这里的情况：中间状态就是结果本身的叫分布型（SUM、COUNT、MAX、MIN）；中间状态是固定几个值的叫代数型（AVG 存和与个数）；中间状态的大小没有上限、必须看到全部输入的叫整体型（中位数、众数）。精确去重的中间状态是整个用户集合，大小随人数增长，实践中和整体型一样难办。

所以第一个问题真正问的是：有没有一个大小固定、又能合并的中间状态。

能撤销：合进去的，还能减出来

第二个问题：近 30 天的结果已经算好，现在要把 30 天前那一天「拿出来」，做得到吗？

• 成交额：减掉那天的成交额。能。

• 平均订单金额：金额之和、订单数各减掉那天的值。能。

• 最大单笔：不能。离开窗口的那天如果恰好是最大值，剩下 29 天的最大值是多少，光凭当前的最大值不知道。

• 下单人数：不能。那天下单的人，有没有在另外 29 天也下过单，光凭人数不知道。

上图就是全文的骨架。下面分成两条路讲：计算模式怎样利用这两个性质，数据结构怎样创造这两个性质。

计算模式一：只算变化的部分

每天重扫 30 个分区，是把「今天的近 30 天」当成一个全新的问题。增量计算换个问法：昨天的答案已经有了，今天变了什么？

累计窗口：只要能合并

「本月至今」的起点不动，每天只多一天：今天的结果 = 昨天的结果 ⊕ 今天的数据（⊕ 表示合并）。四个指标都能这样算，最大单笔取两者较大的那个，下单人数把两个集合或两个 HyperLogLog 合并。

滑动窗口：还要能撤销

「近 30 天」每天进来一天、出去一天：

⊖ 就是撤销。成交额和平均订单金额能撤销，每天只读 3 个分区，这就是《消灭暴力扫描》里的递推公式。最大单笔和下单人数不能撤销，公式到这里就断了。

不能撤销：切成小块，只合并

不能撤销，就不要走「减出来」这条路：每天存一份中间状态，算近 30 天时把窗口里的 30 份合并起来，离开窗口的那天直接丢掉，不需要减。

• 最大单笔：每天存一个日最大值，30 个取最大。《消灭暴力扫描》里用 MAP 保存「日期 → 当天的值」，就是这个做法。

• 下单人数：每天存一份 Bitmap 或 HyperLogLog，30 份合并后再数。

读的仍然是 30 份，但每份是一个城市一天一个值或几 KB，而不是 30 个分区的订单明细。这里有一个容易踩的坑：HyperLogLog 能合并、不能撤销，所以近 30 天的下单人数不能套用「减掉 30 天前那天」的递推公式，只能把 30 份合并。《大数据去重统计》说 HyperLogLog「每天存一份，按需合成周、月」，原因就在这里。

MaxCompute 的渐进计算替你做的也是这件事：第一次运行时产出按天、按月预聚合的中间结果，之后只算当天的增量，再和中间结果合并。

流计算里，撤销叫回撤

流上没有「昨天」和「今天」，只有一条条变化。订单从「已支付」改成「已退款」，Flink SQL 会把它拆成两条消息：撤回旧值（-U）、写入新值（+U）。下游的聚合收到 -U，要做的正是撤销。

成交额收到 -U 直接减。最大单笔呢？Flink 的 MaxWithRetractAggFunction 在状态里为每个出现过的值记一个次数（MapView<值, 次数>），当前最大值被撤回时，从里面找下一个。这和离线「把每天的值都存下来」是同一个办法：不能撤销的聚合，只能多存状态来换。回撤的原理和它带来的副作用，见《Flink SQL 回撤机制与问题》。

计算模式二：把活拆开

分布式计算的前提也是能合并：每个任务算一块，最后合起来。北京的订单占了一半，问题是这一块太大，那就把它再拆开。

两阶段聚合：先局部合并，再全局合并

先按「城市 + 一个桶号」局部聚合，把北京这个大组拆成很多小组；再按城市把局部结果合并：

SELECT city, SUM(part_gmv) AS gmv, MAX(part_max) AS max_amount
FROM (
    SELECT city, SUM(amount) AS part_gmv, MAX(amount) AS part_max
    FROM orders
    WHERE dt > '${bizdate-30d}' AND dt <= '${bizdate}'
    -- 局部：北京被拆成 64 个小组，分给 64 个任务
    GROUP BY city, MOD(HASH_CODE(order_id), 64)
) t
-- 全局：每个城市只剩 64 行要合并
GROUP BY city;

成交额、最大单笔、平均订单金额（存和与个数）都能这样拆，因为它们能合并。Flink SQL 的 Local-Global 聚合（table.optimizer.agg-phase-strategy）是同一个思路，只是局部合并发生在数据发往下游之前。

下单人数不能按订单号随便分桶，否则同一个人落进几个桶，各桶人数一加就重复了。办法是按 user_id 分桶：同一个人只会落进一个桶，桶与桶之间没有交集，这时各桶的人数可以直接相加。注意这并没有让去重变得能合并，而是把数据切成互不相交的几块，集合不相交时大小才能相加。Flink SQL 的 table.optimizer.distinct-agg.split.enabled 自动做的就是这一步，细节见《大数据性能调优》。

切片窗口：每条数据只算一次

风控规则常要「近 1 天，每分钟更新一次」。如果每个窗口单独累加，一条数据属于 1440 个窗口，就要被累加 1440 次。切片把时间切成互不重叠的 1 分钟小段，每条数据只进一个切片，窗口的结果由它包含的切片合并而成。

切片同样只需要能合并：最大单笔没法撤销，照样能用切片。《Flink 规则引擎》里能随时改的窗口就是这样实现的，Flink SQL 的窗口表值函数也是。

数据结构：让中间状态又小又能合并

前两部分都有个前提：要有一个好的中间状态。没有现成的，就设计一个。

• 预聚合表：数仓的汇总层（DWS）本质上就是把中间状态落盘：按城市、按天存好，上层任意周期都从它合并。所以汇总层该存的是金额之和与订单数，而不是平均值；是 HyperLogLog，而不是一个人数。《Life of a Data》里汇总层的 uv=<HLL> 就是这个道理。

• 压缩和概率结构：Bitmap 让精确去重的状态小几十倍，仍然随人数增长；HyperLogLog 用 1%～2% 的误差，换来固定几 KB、能合并的状态。代价是它不能撤销，也算不准交集。原理见《大数据去重统计》。

• 存储层聚合：Fluss 的聚合表在写入时按主键和已有的值合并，把聚合从 Flink 的状态里挪到存储。能挪下去的，也只有能合并的聚合；而它不接受回撤，原因之一就是在写入即可见的模型里，没法安全地撤销。见《Fluss：重塑实时开发》。

还有一类优化和聚合的性质无关，目标却相同，都是少读、少搬：按分区和文件统计跳过数据，按关联键分桶让 Join 不搬数据（Colocate），用 Runtime Filter 先过滤再搬。这些都取决于数据在存储里怎么摆，放在《大数据性能调优》的第二部分讲。

代价：没有白来的优化

总结：拿到一个指标，先问两个问题

回到开头：递推、HyperLogLog、两阶段聚合、切片，是同一件事的不同侧面。拿到一个新指标，可以按这个顺序判断：

1. 它能撤销吗？能，滑动窗口用递推，流上的回撤直接减。

2. 不能撤销，但有小的、能合并的中间状态吗？有，就按时间切片，存中间状态只合并不撤销；倾斜时做两阶段聚合。

3. 中间状态太大，或者根本不能合并？换数据结构：切成不相交的桶让它能相加，换近似结构把状态压小，或者接受多存明细。

4. 都做了还慢？看数据怎么分、怎么搬：倾斜和 Shuffle。

我的判断是，这些优化最终都该由平台根据聚合的性质自动选择，而不是开发者手写。MaxCompute 的渐进计算、Flink SQL 的 Local-Global 和拆分去重开关、基于 DBSP 的 Feldera，都在往这个方向走。即便如此，知道这两个问题仍然有用：平台帮不上的时候，你知道卡在了哪里。

参考资料

• Jim Gray 等，Data Cube: A Relational Aggregation Operator Generalizing Group-By, Cross-Tab, and Sub-Totals，Data Mining and Knowledge Discovery 1(1)，1997。

• Mihai Budiu 等，DBSP: Automatic Incremental View Maintenance for Rich Query Languages，PVLDB 16(7)，2023。

• Flink 文档：性能调优（MiniBatch、Local-Global、Split Distinct）。

• Flink 源码：MaxWithRetractAggFunction。

• MaxCompute 文档：渐进计算。