上周例行巡检数仓计算集群时，一条监控指标异常刺眼：一个平时只要跑两分钟的日级汇总任务，昨天突然跑了近一个半小时，Spark UI 上一个原本平平无奇的 Stage 堆满了上百次 Shuffle Spill（Disk）。点进执行日志一看，某个业务线新增了按“用户客户端设备指纹（device_fingerprint）”进行分组聚合的逻辑。
原本业务只按“省份 + 年龄段 + 操作系统”聚合，总分组数（基数 Cardinality）撑死也就几千个，内存里的哈希表轻轻松松装下。但一引入拥有上千万离散值的高基数字段，执行引擎内部的哈希聚合桶（Hash Aggregation Map）直接被打穿，大量内存对象被频繁溢写到磁盘，I/O 读写队列直接把节点磁盘打死。
在大数据分析、OLAP 引擎与现代分布式计算中，GROUP BY 是使用频率最高却也最容易导致节点内存崩溃的算子。很多人知道它慢，却很少有人真正拆开底层去看：哈希聚合（Hash Aggregate）与流式排序聚合（Streaming / Sort Aggregate）在 CPU 缓存、内存分配器与磁盘 I/O 之间，究竟经历了怎样的极限拉锯？
一、 两种聚合算子的底层解剖
数据库或分布式计算引擎在面对一条包含 GROUP BY 的 SQL 时，物理执行器通常只有两种基础战术路线可走：
【哈希聚合 (Hash Aggregate)】
输入流 (无序) —> [计算 Hash Key] —> 查找内存 HashTable
|– 存在: 累加聚合度量 (In-place Update)
\\– 不存在: 动态扩容插槽 (Rehash / OOM 隐患)
* 特点：无需前序排序，低基数下快如闪电；高基数下内存吞吐崩溃。
【流式/排序聚合 (Streaming / Sort Aggregate)】
输入流 (按 Group 键预先排序) —> 逐行扫描对比当前 Key 与前一行 Key
|– 相同: 累加局部累加器
\\– 变化: 发射上一组结果，清空累加器复用内存！
* 特点：内存开销为 O(1) 常数级！但严重依赖前序排序代价。
1. 哈希聚合的甜蜜与毒药
哈希聚合的核心是利用内存中的哈希表（如 ClickHouse 的 AggregatedDataVariants 或 Spark 的 BytesToBytesMap）。
- 优势：对输入数据没有排序要求，只要基数小，一条记录进来计算哈希值，命中桶位后直接在原地内存对状态变量（如 count++, sum += val）执行累加，时间复杂度理论上是 $O(N)$。
- 死穴：哈希表必须完整常驻内存。一旦分组键基数达到千万级，哈希表不仅占用数十 GB 空间，还会产生大量的哈希冲突与动态扩容重排（Rehash），引发极其严重的 CPU Cache Miss 和内存碎片。
2. 流式聚合的优雅与代价
流式聚合的核心前提是：输入流必须已经按照聚合键排好序。
- 优势：执行器只需要在内存中维护当前一个分组的累加状态（一个整型计数器和一个求和累加器即可）。当读取到下一个不同的分组键时，直接将当前分组的最终值发射（Emit）给下游，并将累加器重置为 0。无论底层数据量是一亿行还是一百亿行，聚合算子的瞬时内存占用恒定保持在几十个字节（$O(1)$）！
- 代价：如果底层数据物理存储本身是无序的，为了使用流式聚合，必须前置执行代价高昂的全量分布式排序（Sort / Shuffle），而排序本身也是耗内存大户。
二、 内存极限压榨实战：如何逼迫引擎切入流式聚合？
既然哈希聚合在高基数下必崩，而无序数据的排序代价又高，工程上的终极优化思路就是：利用列式存储的物理预排序索引，绕过外部排序，直接把流式聚合推至物理扫描层（Scan-Time Streaming Aggregation）。
1. ClickHouse 的物理主键对齐魔法
在 ClickHouse 中，如果我们的查询分组键恰好与表的主键排序前缀（ORDER BY (tenant_id, event_date, client_type)）高度吻合，ClickHouse 可以利用 optimize_aggregation_in_order = 1 这一神级参数：
— 开启按主键排序顺序流式聚合，内存压降 90% 以上
SELECT
tenant_id,
event_date,
count(1) AS pv,
sum(amount) AS total_amt
FROM dwd_events_all
WHERE event_date >= '2026-09-01'
GROUP BY tenant_id, event_date
SETTINGS optimize_aggregation_in_order = 1,
max_threads = 8;
原理在于：由于存储文件（Parts）内部的数据物理上已经是按 (tenant_id, event_date) 连续排好序的，执行引擎无需在内存中维护全量哈希大字典，各个线程只需对当前数据块进行顺序扫描与折叠，流式输出局部结果，然后进行轻量级两路归并，内存占用直接从几十 GB 断崖式下跌到几十 MB。
2. Spark / SQL 引擎下的两阶段盐值打散（Two-Phase Aggregation）
如果高基数无法利用物理排序（例如不得不按全局随机分布的 device_fingerprint 聚合），且不可避免引发单个节点内存倾斜，我们必须在代码层面打破单一哈希大表，采用“加盐局部哈希 + 全局流式规约”的两阶段打散：
— 第一阶段：注入随机前缀（加盐），将单个巨大的哈希桶强制打散至 N 个并行分片
WITH salted_data AS (
SELECT
concat(device_fingerprint, '_', cast(floor(rand() * 16) AS int)) AS salted_key,
device_fingerprint,
amount
FROM ods_raw_access_log
WHERE dt = '2026-10-04'
),
— 第二阶段：局部轻量哈希聚合，大幅减少原始记录条数
partial_agg AS (
SELECT
salted_key,
device_fingerprint,
count(1) AS partial_cnt,
sum(amount) AS partial_sum
FROM salted_data
GROUP BY salted_key, device_fingerprint
)
— 第三阶段：去掉盐值，此时输入规模已收缩几个数量级，全局收敛无压力
SELECT
device_fingerprint,
sum(partial_cnt) AS total_cnt,
sum(partial_sum) AS total_amount
FROM partial_agg
GROUP BY device_fingerprint;
三、 聚合算法自适应选择器设计
在很多自主可控的查询中间件（如自研 ChatBI 接入层）中，我们不能完全依赖黑盒优化器的随机猜测。可以通过元数据驱动构建一个聚合策略预裁决模块：
from enum import Enum
from dataclasses import dataclass
from typing import List
class AggMethod(Enum):
HASH_AGG = "HASH_AGG"
SORT_STREAMING_AGG = "SORT_STREAMING_AGG"
SPILL_TO_DISK_HASH = "SPILL_TO_DISK_HASH"
@dataclass
class QueryContext:
table_name: str
group_by_columns: List[str]
estimated_rows: int
column_ndv: dict # Number of Distinct Values 独立基数估算
table_sort_keys: List[str]
class AggregationOptimizerRule:
MEMORY_LIMIT_BYTES = 4 * 1024 * 1024 * 1024 # 4GB 节点保护水位
BYTES_PER_HASH_ENTRY = 64 # 单个哈希桶开销经验估算
@classmethod
def decide_strategy(cls, ctx: QueryContext) -> AggMethod:
# 1. 估算全量分组基数乘积
total_cardinality = 1
for col in ctx.group_by_columns:
total_cardinality *= ctx.column_ndv.get(col, 1000)
estimated_hash_mem = total_cardinality * cls.BYTES_PER_HASH_ENTRY
# 2. 核心判断：分组键是否为表物理排序键的前缀？
is_prefix_sorted = (
len(ctx.table_sort_keys) >= len(ctx.group_by_columns) and
ctx.table_sort_keys[:len(ctx.group_by_columns)] == ctx.group_by_columns
)
if is_prefix_sorted:
# 物理上有序，毫不犹豫走流式零内存聚合
return AggMethod.SORT_STREAMING_AGG
if estimated_hash_mem < cls.MEMORY_LIMIT_BYTES:
# 内存充裕且基数可控，优先走极速哈希聚合
return AggMethod.HASH_AGG
else:
# 高基数且无序，必须显式限制内存并配置写盘降级
return AggMethod.SPILL_TO_DISK_HASH
