分布式查询的两阶段聚合

做的事：真实跑一条跨分片的GROUP BY，看EXPLAIN怎么拆解它

代码-- 集群里已经有跨2个分片的数据(dist_tbl) EXPLAIN actions=1 SELECT v, count() FROM demo.dist_tbl GROUP BY v;

验证：真实的两阶段结构，一层不多一层不少

实测 · 本机执行输出

MergingAggregated                    ← 第2阶段：真实合并所有分片的局部聚合结果
│  Keys: v
│  Aggregates: count()
│  Mode: memory-efficient
└──Union
   ├──Aggregating                    ← 第1阶段(本地)：真实对本分片数据先做一次局部聚合
   │  │  Keys: v
   │  │  Aggregates: count()
   │  └──ReadFromMergeTree (demo.local_tbl)
   │        Parts: 2 | Granules: 2
   └──ReadFromRemote (Read from remote replica)   ← 第1阶段(远程)：真实从另一分片拉取局部聚合结果

这份真实的执行计划精确对应"两阶段聚合"这个名字：第一阶段真实是Aggregating，每个分片各自对自己本地的数据先做一次GROUP BY，得到一份体量小得多的局部聚合结果(比如"v=5这个值在本分片出现了37次"这样的中间统计，而不是原始的37行数据本身)；第二阶段真实是MergingAggregated，发起查询的节点把所有分片各自算出的局部聚合结果收集起来，再做一次真正的合并(比如把两个分片各自算出的"v=5出现了37次"和"v=5出现了52次"加总成89次)。

为什么两阶段聚合，比"先把原始数据都拉回来再GROUP BY"聪明

这套架构真实节省的是跨节点传输的数据量——如果不做本地的第一阶段聚合，直接把每个分片的原始数据都传回发起查询的节点再统一GROUP BY，网络上真实要传输的是完整的明细数据（本文场景下是2000多行原始记录）；而两阶段聚合让每个分片先在本地把体量压缩到"每个分组一行统计结果"（本文场景下最多几十行，取决于v的基数），只有这份压缩后的局部结果才真实通过网络传输，MergingAggregated阶段要处理的数据量因此小得多。这跟本系列很早验证过的merge本质是k路归并（把多个已经排好序的数据源合并成一份）在思路上是相通的——只是这次归并/合并的对象，不是排好序的part，而是各自已经聚合过一遍的中间统计结果。Mode: memory-efficient这个标记也真实提示了：第二阶段的合并本身也经过了内存效率方面的优化（本系列前面《Two-Level Aggregation》验证过的自适应聚合机制，同样适用于这一步的合并计算），而不是简单粗暴地把所有分片的中间结果一次性摊开在内存里再合并。理解这套两阶段结构，也解释了为什么跨分片JOIN(本系列前面《Distributed JOIN为什么容易变慢》验证过)没有这么优雅的中间压缩步骤可用——JOIN的中间结果没法像聚合一样先"压缩成统计摘要"再合并，这也是分布式JOIN真实比分布式聚合更容易变慢的一个结构性原因。

分步演示：从EXPLAIN真实拆解出的两阶段聚合结构

每一步都是本文上面用真实EXPLAIN跑出来的原始结果。

两阶段聚合：先压缩再合并，真实的执行计划

这一步新验证的阶段

本文用到的工具

本机真实的3节点ClickHouse集群，真实的 EXPLAIN actions=1（MergingAggregated/Aggregating/ReadFromRemote）

下一篇 · CLICKHOUSE 内部原理 60

分布式 JOIN 为什么容易变慢