Connected Component from DFS to BigQuery

去年迁移地图处理 pipeline 的时候，遇到了一段道路连通分量计算的代码。原本这部分的处理用的是 GraphFrame 的图计算库 + GraphFrames 是一个用于 Apache Spark 的软件包，它提供基于 DataFrame 的图功能。它提供 Scala、Java 和 Python 的高级 API： https://graphframes.io/ ，非常简单好用。但迁移后的 BigQuery 的官方文档中并没有图相关的函数，就随便让 Gemini 帮我糊了一个 SQL 实现，跑了一下感觉还行就直接用了。

原本的代码大概是这样的：

import org.apache.spark.sql.functions._
import org.graphframes.GraphFrame

// 准备 node 和 edge
val roads = spark.table("road").select("src", "dst", "oid")
val edges = roads.union(roads.select("dst", "src", "oid").toDF("src", "dst", "oid"))
val nodes = edges.select("src").distinct().union(edges.select("dst").distinct()).toDF("id")

// 计算连通分量
val g = GraphFrame(nodes, edges)
val result = g.connectedComponents.setUseLocalCheckpoints(true).run()

// 将结果与原始道路表连接，得到每条道路所属的连通分量
val road_x_connected_component = result.join(roads, expr("id=dst"))
  .union(result.join(roads, expr("id=src")))
  .select("oid", "component")
  .dropDuplicates()

说实话这个代码真正计算连通分量的地方就只有 GraphFrame 封装好的一个函数，里面究竟干了什么我也不知道。Gemini 给我生成的 SQL 代码我也没太看明白，所以这里我再重新学习一下这个基础图问题的相关算法。

那么首先什么是连通分量呢？

给定一个无向图 $G = (V, E)$ ，其中 $V$ 是节点集合， $E$ 是边集合。一个连通分量是图中一个极大的节点子集，使得子集中的任意两个节点之间都存在一条路径，且无法再加入任何外部节点而保持这一性质。

换句话说，连通分量就是图中彼此”连通”的节点群组。每个节点恰好属于一个连通分量，所有连通分量构成了对整个图的一个划分。

一个直观的例子：假设有 6 个节点和以下 4 条边：

□

一个简单的无向图

1 -- 2
2 -- 3
3 -- 4
5 -- 6

这个图绘制出来就会如右图所示，一共有两个连通分量：{1, 2, 3, 4} 和 {5, 6}。节点 1 可以经由 2、3 到达 4，但无论如何都无法到达 5 或 6。

所以连通分量计算的目标就是为每个节点分配一个标签（通常是该分量中最小的节点 ID），使得同一分量中的节点拥有相同的标签。

经典单机算法：DFS/BFS 与并查集

DFS/BFS 遍历

首先肯定不能绕过的是深度优先搜索（DFS）或广度优先搜索（BFS），感觉一下子梦回大学的数据结构课程。虽然现在上班好多年，已经不能手写代码了，但算法的核心思想还是记得的。这里稍微整理一下思路：

从一个未访问的节点出发，通过 DFS 或 BFS 遍历所有可达节点，将它们标记为同一分量。重复这个过程直到所有节点都被访问。

整体时间复杂度为 $O(V + E)$ ，其中 $V$ 是节点数， $E$ 是边数。对于使用邻接表存储的图，这已经是最优的了。如果图是非常稠密的（ $E \approx V^2$ ），那么 $O(V + E)$ 实际上就等同于 $O(V^2)$ 。

整体上在单机内存能容纳整个图的情况下非常高效，然而它本质上是一个顺序算法。整个遍历过程依赖于前一步的结果，难以并行化。如果图有数十亿个节点和边时，那单机的内存和计算能力就都扛不住了。

并查集（Union-Find / Disjoint Set Union）

Tarjan, R. E. (1975). “Efficiency of a Good But Not Linear Set Union Algorithm.” Journal of the ACM, 22(2), 215–225。

另一个经典方法是基于并查集数据结构。初始时每个节点自成一个集合。然后遍历每条边 $(u, v)$ ，将 $u$ 和 $v$ 所在的集合合并。处理完所有边后，属于同一集合的节点就属于同一连通分量。

并查集支持两种核心操作：Find（找到节点所属集合的代表元素）和 Union（合并两个集合）。通过路径压缩（Path Compression）和按秩合并（Union by Rank）两种优化，每次操作的均摊时间复杂度为 $O(\alpha(n))$ ，其中 $\alpha$ 是反阿克曼函数。

{\displaystyle A(m,n)={\begin{cases} n+1 &\text{if } m=0\\ A(m-1,1) &\text{if } m>0 \text{ and } n=0\\ A(m-1,A(m,n-1)) &\text{if } m>0 \text{ and } n>0\\ \end{cases}} }

一个增长极其缓慢的函数，对所有实际输入规模都可以视为常数。

并查集的理论最优性分析参见 Fredman, M. & Saks, M. (1989). “The Cell Probe Complexity of Dynamic Data Structures.” STOC ‘89。

并查集的优势在于它可以按任意顺序处理边，甚至支持流式处理。但和 DFS/BFS 一样，它本质上仍是单机顺序算法，存在指针追逐问题：当前要访问的内存地址依赖于上一次读取出来的内存地址。所以也不足以处理大量数据。

基于并行计算 PRAM 模型的算法

想让成百上千个处理器同时干活，就必须先回答一个更基础的问题：多个处理器在同一时刻读写同一块内存时，究竟会发生什么？ PRAM 并行随机存取机 (Parallel Random Access Machine) 就是为了回答这个问题而被造出来的模型。

PRAM 的基本假设

PRAM 模型由 Fortune, S. & Wyllie, J. (1978). “Parallelism in Random Access Machines.” STOC ‘78, 114–118 正式提出。系统的教材式介绍可参考 JáJá, J. (1992). An Introduction to Parallel Algorithms。

PRAM 可以理解为把单机的 RAM (Random Access Machine) 模型原样复制若干份，再让它们共享同一块内存：

处理器： $p$ 个处理器 $P_1, \dots, P_p$ ，每个都是一台标准 RAM，拥有自己的局部寄存器，并且知道自己的编号 $i$ 。很多算法就靠这个编号来分工，比如”第 $i$ 个处理器负责第 $i$ 条边”。
共享内存：一块容量无限的全局内存，任何处理器都能在单位时间内访问其中任意一个单元。没有 cache，没有 NUMA，没有网络延迟，也没有带宽上限。
同步执行：所有处理器步调一致（lock-step），每一步分为三个阶段——从共享内存读取、本地计算、向共享内存写回。所有处理器完成当前这一步之后，才会一起进入下一步。

可以看出这个模型极其慷慨：它把”通信”的成本直接抹成了零，只留下”计算需要多少轮”这一个变量。这当然不现实，但好处是让算法设计者可以先专心思考并行性本身的极限：在一个完全没有通信开销的理想世界里，一个问题最少需要几轮才能算完？

Brent 定理：一个需要 $T$ 轮、总工作量为 $W$ 的并行算法，可以在 $p$ 个处理器上用 $O(W/p + T)$ 的时间模拟。这意味着处理器不够用时也不会吃太大的亏，因此算法设计者可以放心地先假设”处理器管够”。参见 Brent, R. P. (1974). “The Parallel Evaluation of General Arithmetic Expressions.” Journal of the ACM, 21(2), 201–206。

为此 PRAM 有两个评价指标：时间 $T$ （轮数）和处理器数 $p$ 。两者的乘积 $W = T \cdot p$ 称为工作量（work），代表所有处理器忙碌时间的总和。如果一个并行算法的 work 与最好的串行算法同阶，就称它是工作最优（work-optimal）的。连通分量的串行代价是 $O(V + E)$ ，所以”用 $O(V + E)$ 个处理器跑 $O(\log n)$ 轮”虽然够快，但 work 达到了 $O((V + E)\log n)$ ，比串行多做了 $\log n$ 倍的无用功——这正是后来 Gazit 等人要改进的地方。

读写冲突：EREW、CREW 与 CRCW

模型里唯一需要认真定义的细节是：当多个处理器在同一步里访问同一个内存单元时，该怎么办？ 对”并发读”和”并发写”分别选择允许或禁止，就得到了 PRAM 的几个变体：

模型	全称	同时读	同时写
EREW	Exclusive Read Exclusive Write	禁止	禁止
CREW	Concurrent Read Exclusive Write	允许	禁止
CRCW	Concurrent Read Concurrent Write	允许	允许

并发读还算好理解，无非是把同一个值广播给多个处理器理论上还存在 ERCW（禁止并发读但允许并发写）这一格，但既然写冲突这个更难的问题都解决了，再去限制读就没什么意义，因此几乎无人研究。；真正麻烦的是并发写：如果 100 个处理器同时往同一个单元写入 100 个不同的值，最后留下的应该是哪一个？于是 CRCW 内部又分成了几个流派：

Common（一致写）：只有当所有写入者要写的值完全相同时才合法，否则就算算法本身有 bug。
Arbitrary（任意写）：随便谁赢都行，算法必须保证无论谁赢结果都正确。
Priority（优先写）：编号最小的处理器赢。
Combining（归约写）：由硬件把所有写入值做一次归约（取 min / max / 求和）之后再写入。

这几种模型的能力依次递增： $\text{EREW} \le \text{CREW} \le \text{CRCW(Common)} \le \text{Arbitrary} \le \text{Priority}$ 。不过差距并没有想象中那么大——任何 CRCW 算法都可以在 EREW PRAM 上模拟，代价是 $O(\log p)$ 的减速，做法是用排序和前缀和把并发访问”串行化”掉。

正因如此，在 CRCW 上做出 $O(\log n)$ 和在 EREW 上做出 $O(\log n)$ 是两件难度完全不同的事情：前者是 1982 年 Shiloach 和 Vishkin 的成果，后者要等到十年之后 Karger 等人的随机化算法。这也是为什么后文的表格里，每个算法都要专门标注它所依赖的模型。

而对连通分量来说，CRCW 尤其自然：算法的核心动作是”把邻居们的标签取最小值”，在 Combining CRCW 上这直接就是一条并发 min-write 指令；即便退到 Arbitrary CRCW，“任意一个邻居胜出”通常也只影响收敛的轮数，不影响最终结果的正确性。这就是这一时期的算法几乎都默认站在 CRCW 上的原因。

Shiloach-Vishkin 算法（1982）

Shiloach, Y. & Vishkin, U. (1982). “An O(log n) Parallel Connectivity Algorithm.” Journal of Algorithms, 3(1), 57–67。

1982 年，Yossi Shiloach 和 Uzi Vishkin 在 Journal of Algorithms 上发表了一个具有开创性意义的并行连通分量算法。它在 CRCW PRAM 上运行， $O(\log n)$ 时间， $O(V + E)$ 个处理器。

数据结构：父指针森林

算法的全部状态就是一个数组 $P$ ： $P[v]$ 是节点 $v$ 的父节点，根节点 $r$ 满足 $P[r] = r$ 。初始时 $P[v] = v$ ，每个节点自成一棵单点树。

这里有个容易被跳过但很关键的设计选择：为什么要维护一片树，而不是像并查集那样直接给每个节点存一个标签？ 因为标签是”平”的，要让一个标签走到 $k$ 跳之外的节点，就得传播 $k$ 轮；而树是”有结构”的，一次指针跳跃就能把所有节点到根的距离减半。 $O(\log n)$ 这个结果，本质上就是从这个减半里来的。

算法维持的不变式是： $P$ 始终构成一片有根树的森林，同一棵树里的节点必定属于同一个连通分量。算法结束时每棵树都被压成深度为 1 的星，根的编号就是这个分量的标签。

三个操作

□

一轮迭代：钩连让树变高，指针跳跃再把它压平。两者交替，森林在 O(log n) 轮内塌缩成星。

① 条件钩连（conditional hooking）。 对每条边 $(u, v)$ 并行地看：如果 $u$ 的父亲已经是一棵树的根，而 $v$ 那边的父亲编号更小，就把 $u$ 这棵树的根挂到 $v$ 的父亲下面。用一句话说就是——一条边如果连着两棵不同的树，就让编号大的那棵认编号小的那棵当爹。

这一步正是需要 CRCW 的地方：可能有很多条边同时想改写同一个 $P[\text{root}]$ 。在 Arbitrary CRCW 下随便谁赢都行，因为无论谁赢，写进去的值都比原来的根小，正确性不受影响，只影响收敛的快慢。

② 指针跳跃（shortcutting）。 所有节点并行执行 $P[v] \leftarrow P[P[v]]$ ，也就是”直接认祖父当爹”。每执行一次，每个节点到根的距离减半，树高从 $h$ 掉到 $\lceil h/2 \rceil$ 。

③ 停滞钩连（stagnant hooking）。 只有①还不够。设想一棵树的根 $r$ ，它周围所有的边都通向编号比它大的树——条件" $P[v] < P[u]$ "处处不成立，于是这一轮谁也没动，算法卡住了，但活儿还没干完。所以 SV 补了第二次钩连：对于在①里没能挂出去的根，允许它无条件地挂到任意一个邻居的树上。

// P[v] 是 v 的父节点；根节点满足 P[r] = r
parallel for v ∈ V:
    P[v] ← v

repeat until P 不再变化:

    // ① 条件钩连：把编号大的根挂到编号小的那边
    parallel for (u, v) ∈ E:
        if P[P[u]] = P[u] and P[v] < P[u]:
            P[P[u]] ← P[v]           // 并发写：Arbitrary CRCW，谁赢都对

    // ② 停滞钩连：给①里谁也没挂上的根一次无条件的机会
    parallel for (u, v) ∈ E:
        if P[P[u]] = P[u] and P[u] 在①中未被写过:
            P[P[u]] ← P[v]

    // ③ 指针跳跃：直接认祖父，树高减半
    parallel for v ∈ V:
        P[v] ← P[P[v]]

右图就是一轮迭代的样子。初始有两棵树，根分别是 2 和 3，中间隔着一条图上的边 $(5, 4)$ 。钩连这一步，边 $(4, 5)$ 让 3 这棵树的根挂到了 2 上——注意此时树变深了，4 和 9 离根有 2 跳。紧接着的指针跳跃把这个深度重新压掉：4 和 9 原本指向 3，而 $P[3]$ 已经是 2，于是它们一步就直接指向了 2。

一高一压，正是这个算法的全部节奏。 钩连负责合并，它的副作用是把树拉长；指针跳跃负责把拉长的部分收回来，让下一轮的钩连仍然发生在”根与根”之间。原文用一个关于森林结构的势函数证明了：无论输入图长什么样， $O(\log n)$ 轮之后一定收敛。

和并查集的关系

看到这里应该会觉得眼熟：SV 其实就是并查集的并行版本。

并查集（串行）	Shiloach-Vishkin（并行）
Union：合并两个集合	Hooking：把一棵树挂到另一棵树上
路径压缩：Find 时顺手把路径拍平	Shortcutting：所有节点同时认祖父
一条边一条边地处理	所有边同时处理

差别全在最后一行。串行地处理边，天然不会有冲突；一次性并行处理所有边，就必须额外发明规则来避免成环和停滞——①和③这两条别扭的规则，全都是为并行付的税。

后续的 PRAM 改进

Awerbuch, B. & Shiloach, Y. (1987). “New Connectivity and MSF Algorithms for Shuffle-Exchange Network and PRAM.” IEEE Transactions on Computers, C-36(10), 1258–1263。

Johnson, D. B. & Metaxas, P. (1991). “Connected Components in $O(\log^{3/2}|V|)$ Parallel Time for the CREW PRAM.” FOCS ‘91, 688–697。

SV 之后的十年，PRAM 上的改进基本沿着两条线走：把 work 降到最优，以及在更弱的模型上做到同样快。

Awerbuch & Shiloach (1987) 简化了钩连的判定规则，不再需要显式记录”谁在①里被写过”，把三步压成了一套统一的规则，实现起来简单得多。后来几乎所有工程实现参照的都是这个版本。

Gazit, H. (1991). “An Optimal Randomized Parallel Algorithm for Finding Connected Components in a Graph.” SIAM J. Comput., 20(6), 1046–1067。

Karger, D. R., Nisan, N. & Parnas, M. (1992). “Fast Connected Components Algorithms for the EREW PRAM.” SPAA ‘92, 373–381。期刊版：SIAM J. Comput., 28(3), 1021–1034, 1999。

Gazit (1991) 解决了 work 的问题。SV 用 $O(V+E)$ 个处理器跑 $O(\log n)$ 轮，work 是 $O((V+E)\log n)$ ，比串行多做了 $\log n$ 倍。Gazit 用随机采样先把图收缩一轮，做到 $O(\log n)$ 时间 + $O((V + E)/\log n)$ 个处理器，work 与串行同阶——这是工作最优的。
Johnson & Metaxas (1991) 在禁止并发写的 CREW 上给出了 $O(\log^{3/2} n)$ 的确定性算法。
Karger, Nisan & Parnas (1992/1999) 啃下了最硬的 EREW：连”并发读”都不许，min-write 这种一步到位的动作必须靠显式构造广播树来模拟。他们给出了随机化 $O(\log n)$ 和确定性 $O(\log^{1.5} n)$ 的算法。

Eppstein, D. & Galil, Z. (1988). “Parallel Algorithmic Techniques for Combinatorial Computation.” Annual Review of Computer Science, 3, 233–283。

然而这些成果几乎都停留在纸面上。正如 Eppstein 和 Galil 所指出的，PRAM 模型”常被理论计算机科学家使用，却较少被实际并行机器的构建者采用”。原因下一节就会讲清楚。

从 PRAM 到 MapReduce：模型是怎么迁移的

三条假设，同时失效

回头看 PRAM 的三条基本假设，放到一个真实的集群上，没有一条站得住：

“任意地址、单位时间” —— 一次本地内存访问大约 100 纳秒，一次跨机器的网络往返在几百微秒到毫秒量级，差了三到四个数量级。这不是可以塞进大 O 里的常数因子，它会直接改变哪个算法更优的判断。

“lock-step 同步” —— PRAM 假设所有处理器齐步走，而现实里做一次全局同步本身就是一笔可观的开销，何况集群里总有掉队的慢节点。

“没有失败” —— PRAM 里压根没有”机器挂掉”这个概念。而在几千台机器的规模上，任务失败是每分钟都在发生的常态。

Valiant, L. G. (1990). “A Bridging Model for Parallel Computation.” Communications of the ACM, 33(8), 103–111。Valiant 用的词是 “bridging model”——一座架在硬件和算法之间的桥。

中间站：BSP

第一次认真给通信标价的，是 Valiant 在 1990 年提出的 BSP（Bulk Synchronous Parallel）。BSP 把计算切成一连串超步（superstep），每个超步依次做三件事：本地计算、收发消息、全局同步。一个超步的代价被明码标价为

w + g \cdot h + L

其中 $w$ 是本地计算量， $h$ 是这一步收发的消息条数， $g$ 是网络的单位消息成本， $L$ 是同步的固定开销。PRAM 里那个被抹成零的东西，在这里终于有了价格标签。Google 后来的图计算系统 Pregel，用的就是 BSP。

MapReduce：把通信原语砍到只剩一个

Dean, J. & Ghemawat, S. (2004). “MapReduce: Simplified Data Processing on Large Clusters.” OSDI ‘04, 137–150。

2004 年的 MapReduce 比 BSP 还要激进：它干脆规定，唯一的通信方式就是”按 key 分组”。

map:      (k, v)      →  list(k', v')
shuffle:  按 k' 把所有记录分到同一个 reducer          ← 唯一的通信原语
reduce:   (k', [v'])  →  输出

这个限制看着很苛刻，但它换回来的东西非常值钱：因为通信模式是框架规定死的，框架就能自动完成分区、调度、失败重试和落盘。写算法的人不用再管容错——而这恰恰是 PRAM 和 BSP 都不管的部分。

从 PRAM 到 BSP 再到 MapReduce 的计算模型演进 — 三代并行模型。往右走一步，模型对程序员的限制就更多一分，换来的是对真实机器的描述更准一分。

形式化的桥：MRC 模型

Karloff, H., Suri, S. & Vassilvitskii, S. (2010). “A Model of Computation for MapReduce.” SODA ‘10, 938–948。文中的 Vassilvitskii 也是四年后那篇交替算法论文的作者之一。

那么 PRAM 上积累了三十年的算法，究竟能不能搬到 MapReduce 上？2010 年，Karloff、Suri 和 Vassilvitskii 给了一个漂亮的回答。他们提出 MRC 模型把 MapReduce 形式化，约束有三条：单台机器的内存是 $O(n^{1-\epsilon})$ （装不下整个输入）、机器台数是 $O(n^{1-\epsilon})$ 、以及同一个 key 对应的所有数据必须能装进一台机器。跑 $O(\log^i n)$ 轮的问题被归入 $\mathcal{MRC}^i$ 类。

然后他们证明了这样一条定理：任何使用 $O(n^{2-2\epsilon})$ 总内存、跑 $t$ 步的 CREW PRAM 算法，都可以用 $O(t)$ 轮 MapReduce 模拟出来。

这就是那座桥。它意味着 PRAM 上的 $O(\log n)$ 结果，理论上自动送你一个 $O(\log n)$ 轮的 MapReduce 算法。顺带一提，定理里是 CREW 而不是 CRCW：并发读可以靠 shuffle 把同一个值广播给多个 reducer 来实现，而并发写的冲突消解在 MapReduce 里没有直接对应物，只能用 reduce 端的归约去凑。

新的代价模型：多出来的那一维

但”能模拟”和”划算”是两码事。一轮 MapReduce 的固定开销包括作业启动、一次全量 shuffle、以及把中间结果写回磁盘——在 Hadoop 时代还得写三副本。这就把代价模型彻底改写了：

	PRAM	MapReduce
优化目标	轮数 $T$ 、处理器数 $p$	轮数、每轮的数据量
一轮的代价	一次内存访问	一次全量 shuffle + 落盘
看不见的东西	通信、容错	——

多出来的那一维——每轮要搬动多少数据——是 PRAM 完全看不见的。后面会看到，Hash-to-All 的轮数漂亮得很，却因为这一维直接出局。

更要命的是，把 SV 的操作逐条翻译过去之后，账单长这样：

PRAM 里的动作	MapReduce 里的对应	代价
读 $P[v]$	map 端读一条记录	几乎免费
并发 min-write	reduce 端 `GROUP BY` + `MIN`	搭一轮作业的便车
指针跳跃 $P[P[v]]$	自连接 JOIN	一整轮作业
lock-step 同步	作业结束时的隐式 barrier	启动 + 落盘开销

最贵的那一行，恰恰是 SV 用得最凶的操作。于是这之后的所有算法，本质上都在琢磨同一个问题：怎样用更少的 JOIN，做完同样多的指针跳跃。

大数据时代：MapReduce 上的连通分量（2009–2014）

接下来登场的每一个算法，形态上都长得一模一样：图就是一张边表，算法就是一个”把边表变成边表”的作业，反复跑到不动点为止。区别只在于每一轮怎么改写这张表，以及要跑多少轮。

Pegasus / Hash-Min（2009）

Kang, U., Tsourakakis, C. E. & Faloutsos, C. (2009). “PEGASUS: A Peta-Scale Graph Mining System — Implementation and Observations.” Ninth IEEE International Conference on Data Mining (ICDM)。

2009 年的 Pegasus 是第一个跑在 Hadoop 上的 PB 级图挖掘库。它的核心洞见是：一大批看起来毫不相干的图算法——PageRank、直径估计、连通分量——其实都是同一件事的变体，即反复做矩阵-向量乘法。

于是他们把这件事抽象成一个原语 GIM-V（Generalized Iterated Matrix-Vector Multiplication）：把标准矩阵乘法 $\sum_j M_{ij} v_j$ 里的”乘”和”加”换成两个可自定义的函数，再加一个把结果写回的 assign 函数。连通分量就是在 $(\min, \times)$ 这个半环上做迭代：

v_i^{\text{new}} = \min\Big(v_i,\ \min_{j \in N(i)} v_j\Big)

翻译成 MapReduce 就是两行：

// 状态：每个节点 v 的当前标签 L[v]，初始 L[v] = v
map(v, L[v], N(v)):
    emit(v, L[v])                      // 保留自己的标签
    for u ∈ N(v): emit(u, L[v])        // 把标签发给每个邻居

reduce(v, labels):
    L[v] ← min(labels)                 // 取收到的最小值

这就是后来被称为 Hash-Min 的算法：每轮把自己的标签发给所有邻居，再从收到的标签里取最小值。每轮的通信量是干净的 $O(V + E)$ 。

问题出在轮数上。标签每轮只能前进一跳：

Hash-Min 在路径图上的执行过程，标签每轮前进一跳 — 最小标签 1 像涨潮一样，一轮往前推一格。图的直径有多大，就要跑多少轮。

所以 Hash-Min 的轮数是 $O(d)$ ， $d$ 是图的直径。对社交网络来说这没什么问题——六度分隔理论下 $d$ 通常小于 20，跑二十来轮就收敛了，这也正是 Pegasus 当年测的那类图。

但换成道路网就完全是另一回事。这恰好是我开头那个场景：道路的连通分量图是极度稀疏的近平面图，一条国道可能串起几万个路口，直径轻轻松松上万。在这种图上跑 Hash-Min，等于要提交上万轮 MapReduce 作业——每轮都是一次全量 shuffle 加落盘。这个算法在这里是彻底不可用的。

Rastogi 等人的统一框架（2013）

Rastogi, V., Machanavajjhala, A., Chitnis, L. & Das Sarma, A. (2013). “Finding Connected Components in Map-Reduce in Logarithmic Rounds.” IEEE 29th International Conference on Data Engineering (ICDE), 50–61。论文全文可在 arxiv.org/abs/1203.5387 获取。

2013 年 Rastogi 等人做了一件很有价值的事：他们没有直接提出”又一个算法”，而是先画出了整个设计空间。

他们的框架是这样的：每个节点 $v$ 维护一个簇 $C_v$ （一个节点集合，初始是 $v$ 和它的邻居）。每一轮迭代做两件事——用一个哈希函数 $h$ 决定”把 $C_v$ 的哪些部分发给谁”，再用一个合并函数 $m$ 把每个节点收到的东西并起来。换一对 $(h, m)$ ，就得到一个不同的算法。

Hash-Min：只把 $\min(C_v)$ 发给邻居。通信量最省， $O(d)$ 轮。
Hash-to-All：把整个 $C_v$ 发给 $C_v$ 里的每一个成员。这样每个节点每轮都能看到”邻居的邻居的全部”，簇的半径每轮翻倍，于是只要 $O(\log d)$ 轮。代价是每轮通信量高达 $O(n \cdot |V| + |E|)$ ——簇一大就直接爆炸。
Hash-to-Min：一个漂亮的折中。把整个 $C_v$ 只发给簇内编号最小的那个节点，其余成员只收到一个 ID。

Hash-to-All 与 Hash-to-Min 的消息模式对比 — 同样一个 5 个节点的簇：左边 20 条消息，右边 8 条。Hash-to-Min 让最小节点当集散中心，把 n² 降到 n。

Hash-to-Min 为什么还能保持快？因为整簇信息全都汇聚到了最小节点那里，簇的合并在最小节点处一次性完成，不需要人人都知道人人。论文证明了它在路径图上 $O(\log n)$ 轮收敛，一般图上则是猜想。

Hash-Greater-to-Min：把簇只发给编号大于自己的那部分成员。这实际上是 SV 算法的一次高效 MapReduce 移植，并且可以严格证明 $3\log n$ 轮收敛、每轮通信量 $2(|V| + |E|)$ 。这是第一个同时做到对数轮数和线性通信量的 MapReduce 连通分量算法。

算法	迭代轮数	每轮通信量
Hash-Min（即 Pegasus）	O(d)	O(\|V\| + \|E\|)
Hash-to-All	O(log d)	O(n·\|V\| + \|E\|)
Hash-to-Min	路径图 O(log n)，一般图为猜想	实测接近线性
Hash-Greater-to-Min	3 log n	2(\|V\| + \|E\|)

其中 $n$ 是最大连通分量的节点数， $d$ 是图的直径。这张表就是上一节说的那两个维度：左边一列是轮数，右边一列是数据量，好算法要同时把两列都摁住。

CC-MR（2012）

Seidl, T., Boden, B. & Fries, S. (2012). “CC-MR – Finding Connected Components in Huge Graphs with MapReduce.” ECML PKDD 2012, LNCS 7523, 458–473。

时间上稍早一点，Seidl、Boden 和 Fries 提出的 CC-MR 走的是另一条路：节点中心。

每个 reducer 拿到一个节点 $v$ 和它当前的邻接表 $N(v)$ ，令 $m = \min(N(v) \cup \{v\})$ ，然后把邻域里所有比 $m$ 大的节点统统改挂到 $m$ 上。也就是说，每一轮每个节点都在自己的邻域里做一次”局部星形化”。

它真正的巧思在工程上：CC-MR 利用了 MapReduce 的二次排序（secondary sorting），让 reducer 拿到的 value 迭代器第一个元素就是最小的邻居。这样就不必把整个邻接表读进内存再求最小值，可以完全流式地处理。在幂律分布的真实图里，总有那么几个度数几百万的超级节点，能不能流式处理它们，直接决定了作业会不会 OOM。

现在回头看，CC-MR 其实已经非常接近下一节的主角了——“把邻域里比最小值大的节点全改挂到最小值”，这句话几乎就是 Large-Star 的定义。差的那一步是：Kiveris 等人把这个动作拆成了两个方向，并且给出了收敛性证明。

Alternating Algorithm：Large-Star + Small-Star（2014）

Kiveris, R., Lattanzi, S., Mirrokni, V., Rastogi, V. & Vassilvitskii, S. (2014). “Connected Components in MapReduce and Beyond.” Proceedings of the ACM Symposium on Cloud Computing (SoCC), 1–13。论文可在 ACM Digital Library 获取。

2014 年，来自 Google 的 Kiveris、Lattanzi、Mirrokni、Rastogi 和 Vassilvitskii 发表了 “Connected Components in MapReduce and Beyond”，提出了交替算法（Alternating Algorithm）。这是本文叙述的终点。

状态就是一个有向边集

先说清楚算法在操作什么。它的全部状态是一个有向边集， $(u, v)$ 读作” $u$ 指向 $v$ “。初始时把无向图的每条边双向展开放进去。算法每一轮把这个边集重写一遍，最终收敛到”每个节点直接指向自己所在分量的最小编号”。

对每个节点 $v$ ，记它当前的邻居集合为 $N(v)$ ，记 $m_v = \min\big(N(v) \cup \{v\}\big)$ ，也就是邻域（含自己）里的最小编号。两个操作都只做一件事：把某些节点改挂到 $m_v$ 上。区别只在于”某些”是哪些：

Large-Star( $v$ )： 把 $N(v)$ 中所有严格大于 $v$ 的邻居改挂到 $m_v$ 。
Small-Star( $v$ )： 把 $N(v) \cup \{v\}$ 中所有不大于 $v$ 的邻居（包括 $v$ 自己）改挂到 $m_v$ 。

map(边 (u, v)):
    emit(u, v);  emit(v, u)                    // 双向展开成邻接表

reduce(v, N(v)):                               // Large-Star
    m ← min(N(v) ∪ {v})
    for u ∈ N(v) where u > v:      输出边 (u, m)

reduce(v, N(v)):                               // Small-Star
    m ← min(N(v) ∪ {v})
    for u ∈ (N(v) ∪ {v}) where u ≤ v:  输出边 (u, m)

每个操作恰好是一轮 MapReduce：map 端双向展开，reduce 端按节点分组、求 min、重写边。没有任何多余的 JOIN。

跑一遍看看

□

两轮交替：Large-Star 让 5 同时够到 2 和 3，Small-Star 随即在 5 的邻域里把 3 并给 2。（自环边已省略。）

右图用的是文章开头那个图的变体：节点 $\{2,3,4,5,7,9\}$ ，边 $2\text{–}5,\ 2\text{–}7,\ 3\text{–}4,\ 3\text{–}9,\ 4\text{–}5$ 。图里有两个”局部最小值” 2 和 3，它们之间隔着 $4\text{–}5$ 这条边。

第一轮 Large-Star。 在 $v = 4$ 这个 reducer 里， $N(4) \cup \{4\} = \{3,4,5\}$ ，最小是 3，而 5 比 4 大，于是边 $(5, 4)$ 被改写成了 $(5, 3)$ 。与此同时 $v=2$ 那边把 5 和 7 挂到了 2 上。结果就是——节点 5 现在同时指向 2 和 3。这一步的意义在于：两个原本隔了两跳的局部最小值，被拉到了同一个节点的邻域里。

第一轮 Small-Star。 现在轮到 $v = 5$ 这个 reducer： $N(5) \cup \{5\} = \{2,3,5\}$ ，最小是 2，而 3 不大于 5，于是 3 被挂到了 2 下面。两个局部最小值在这里合并了。

第二轮 Large-Star 把剩下的 4 和 9 从 3 那里直接拉到 2，边集不再变化，收敛。总共两轮。

为什么必须两个一起用

这是整个算法里最有意思的地方。我把两个操作分别单独跑了一遍，结果是两个都会停在错误的答案上：

只做 Large-Star，第一轮之后边集变成 $\{(4,3), (5,2), (5,3), (7,2), (9,3)\}$ ，再做多少轮都一模一样——到了不动点，可 2 和 3 仍然是两个根，答案是错的。原因是 Large-Star 只改写”比 $v$ 大的邻居”，而一个局部最小值自己的指针永远不会在自己的邻域里被改动，于是两个局部最小值之间谁也不肯让步。

只做 Small-Star，会停在 $\{(2,2), (3,2), (3,3), (4,2), (5,2), (7,2), (9,3)\}$ ：9 还挂在 3 上，3 又挂在 2 上，树没被压平，也不是最终答案。原因是 Small-Star 只改写”不大于 $v$ 的邻居”，它擅长把局部最小值并起来，却没本事把远处的大节点拉过来。

这个”一拉一并”的分工，和 Shiloach-Vishkin 里”钩连拉长、跳跃压平”的节奏是同一件事的两种写法。三十二年过去，范式没变，变的是每一步的价格。

交替执行，两者刚好互补：Large-Star 负责把大节点拉到局部最小值身边，顺带制造出”同时够到两个局部最小值”的节点；Small-Star 紧接着在这些节点的邻域里，把两个局部最小值合并掉。一拉一并，缺了任何一半都走不到底。

收敛有多快

论文证明了交替算法在 $O(\log^2 n)$ 轮内收敛，并猜想真实的界是 $O(\log n)$ ，实验数据也一直支持这个猜想。我自己在路径图上验了一下——路径图正是 Hash-Min 的最坏情况：

路径长度 $n$	Hash-Min 轮数	交替算法轮数
8	7	3
16	15	4
32	31	5
64	63	6
128	127	7

右边一列干净利落地就是 $\log_2 n$ 。这也解释了为什么它能处理道路网这种大直径的图，而 Hash-Min 不行。

工程上的事

轮数只是账单的一半，另一半是数据量。Large-Star 之后边集里会出现重复边，必须去重，否则几轮下来数据就膨胀了——这也是论文里反复强调”每轮数据量保持线性”的原因。此外他们还做了两个工程优化：用**分布式哈希表（DHT）**把已经收敛的小分量直接摘出去，只对大分量继续迭代。最终的实测是：纯 MapReduce 实现比当时最好的算法快 3 到 15 倍，DHT 增强版快 10 到 30 倍，能扛住数千亿条边的图。

落回 SQL

BigFunctions 的 connected_components 文档：unytics.io/bigfunctions/bigfunctions/connected_components/

绕了一大圈，终于可以回答开头那个问题了：为什么这个算法特别适合 BigQuery？

因为它的每一步——“找邻居”、“取最小值”、“改挂边”——分别就是 SQL 里的 JOIN、GROUP BY ... MIN、SELECT。一轮 Large-Star 写出来是这样（示意，省略了终止判断）：

-- edges(src, dst) 是当前的有向边集
WITH nbr AS (                                   -- 双向展开成邻接表
  SELECT src AS v, dst AS u FROM edges
  UNION ALL
  SELECT dst AS v, src AS u FROM edges
),
mins AS (                                       -- 每个节点邻域（含自己）的最小值 m
  SELECT v, LEAST(MIN(u), v) AS m
  FROM nbr GROUP BY v
)
SELECT DISTINCT n.u AS src, mins.m AS dst       -- 把更大的邻居改挂到 m
FROM nbr AS n JOIN mins ON n.v = mins.v
WHERE n.u > n.v

Small-Star 只需要把最后的 n.u > n.v 换成 n.u <= n.v，并把 $v$ 自己也加进邻域。整个算法就是这两段 SQL 交替跑到边集不再变化为止——不需要把数据导出到任何专门的图处理系统。

BigFunctions 的文档提到，它的实现每轮会持久化中间结果，整体成本大约相当于对输入表做 15 到 30 次扫描。考虑到输入表只有两列，这个价格通常是可以接受的。

回到最开始

GraphFrames 把这个实现叫作 “Two Phase” 或 “big star / small star”，在文档里直接引用了 Kiveris 等人这篇论文。它同时还保留了一个 Pregel 版本的实现，那个就是每轮推进一跳、轮数等于图直径的 Hash-Min。

最后一个意外的收尾：我去翻了 GraphFrames 的文档才发现，那个我一直当黑箱用的 g.connectedComponents.run()，默认走的实现正是 Large-Star / Small-Star 交替算法。

所以绕了这么一大圈——Spark 上那行我没看懂的 API 调用，和 Gemini 帮我糊出来的那段 BigQuery SQL，底下跑的其实是 2014 年那篇论文里的同一个算法。

后续发展（2014 年之后）

Google 2014 年论文之后，连通分量计算的研究并没有停止。后续的工作主要集中在以下几个方向：

进一步优化 MapReduce 实现

Lulli, A., Carlini, E., Dazzi, P., Lucchese, C. & Ricci, L. (2015). “Cracker: Crumbling Large Graphs into Connected Components.” IEEE ISCC 2015, 574–581。

Park, H.-M., Park, N., Myaeng, S.-H. & Kang, U. (2020). “PACC: Large Scale Connected Component Computation on Hadoop and Spark.” PLOS ONE, 15(3), e0229936。

Kim, C., Han, C. & Park, H.-M. (2022). “UniCon: A Unified Star-Operation to Efficiently Find Connected Components on a Cluster of Commodity Hardware.” PLOS ONE, 17(11), e0277527。

Cracker（Lulli et al., 2015）：提出了一种新的分布式算法，通过将大图”碾碎”成小的连通片段来加速计算。
PACC（Park et al., 2020）：提出了分区感知（Partition-Aware）的连通分量算法，通过两步处理（分区 + 计算）、边过滤和草图（sketching）三项技术，在实际数据集上比之前的最优算法快了最多 10.7 倍。
UniCon（Kim et al., 2022）：将 Large-Star 和 Small-Star 合并为单一的 UniStar 操作，进一步简化算法并优化数据膨胀问题，成功处理了拥有 1290 亿条边的超大图。

面向 SQL/MPP 数据库的算法

Bögeholz, H., Brand, M. & Todor, R.-A. (2019). “In-database Connected Component Analysis.” arXiv:1802.09478v2。论文可在 arxiv.org/abs/1802.09478 获取。

2019 年，Bögeholz、Brand 和 Todor 发表了 “In-database Connected Component Analysis”，专门针对 MPP 关系数据库设计了 Randomised Contraction 算法。该算法的核心思想是：通过随机化的图收缩，每一步将图缩小到原来的一个常数比例，经过对数轮次后图缩小到只剩孤立点，这些孤立点就代表了各个连通分量。

与前述算法不同，Randomised Contraction 使用随机化来避免最坏情况，保证对任意输入图都能在期望 O(log |V|) 轮 SQL 查询内完成，且空间需求严格线性。实验表明，它在 MPP 数据库中的表现优于其他领先的连通分量算法。

基于线性代数的分布式实现

Zhang, Y., Azad, A. & Buluç, A. (2020). “Parallel Algorithms for Finding Connected Components Using Linear Algebra.” Journal of Parallel and Distributed Computing, 144, 14–27。论文可在 ScienceDirect 获取。

Zhang、Azad 和 Buluç 基于 Shiloach-Vishkin 算法，设计了用稀疏线性代数原语实现的 LACC 和 FastSV 算法，可以在 GraphBLAS 框架上高效运行。在 Cray XC40 超级计算机上，这些算法扩展到了 4096 个节点（262,144 个核心），处理超过 500 亿条边的图。

演进脉络总结

下表总结了连通分量计算算法的关键里程碑：

年代	计算模型	代表算法/工作	核心贡献
经典	单机	DFS/BFS	O(V+E)，最直观但无法并行
经典	单机	Union-Find (Tarjan, 1975)	近常数均摊时间，支持流式处理
1982	CRCW PRAM	Shiloach-Vishkin	O(log n) 并行算法，引入 Hooking + Shortcutting 范式
1990		BSP (Valiant)	第一次给通信明码标价：w + g·h + L
1991	CRCW PRAM	Gazit	最优随机化 PRAM 算法
1992	EREW PRAM	Karger-Nisan-Parnas	在更严格的 EREW PRAM 模型上达到随机化 O(log n)
2004		MapReduce (Dean & Ghemawat)	通信原语砍到只剩「按 key 分组」，换来自动容错
2009	MapReduce	Pegasus / Hash-Min	首个 PB 级 MapReduce 图挖掘系统，O(d) 轮
2010		MRC (Karloff-Suri-Vassilvitskii)	证明 CREW PRAM 可被 O(t) 轮 MapReduce 模拟
2012	MapReduce	CC-MR	节点中心 + 二次排序，可流式处理超级节点
2013	MapReduce	Hash-to-Min / Hash-Greater-to-Min	首个对数轮+线性通信的 MapReduce 算法
2014	MapReduce	Alternating Algorithm (Kiveris et al.)	Large-Star + Small-Star，O(log²n) 收敛，工业级性能
2015+		Cracker, PACC, UniCon 等	数据膨胀优化、负载均衡、超大规模图
2019	SQL/MPP	Randomised Contraction	纯 SQL 实现，随机化保证 O(log n) 轮

启示与思考

回顾这段演进历程，有几个值得注意的趋势：

理论与实践的鸿沟。 PRAM 上的算法在 1980–1990 年代就已经达到了理论最优，但直到 2010 年代才被成功迁移到实际的分布式系统中。将共享内存算法转化为消息传递范式的分布式算法，远非简单的代码翻译。

简洁设计的力量。 Alternating Algorithm 的成功很大程度上来自于它的简洁性——两个操作交替执行，每个操作都可以用一条 SQL 查询表达。这种简洁性使得它能够在各种不同的计算平台（MapReduce、BSP、SQL 数据库）上轻松实现。

“够用”的收敛保证。 虽然 O(log²n) 的收敛界不如理论最优的 O(log n)，但在实际应用中这个差距几乎不可感知。实验数据持续表明算法以 O(log n) 的速度收敛，这个猜想至今仍然开放。

数据基础设施决定了算法选择。 BigFunctions 选择 Alternating Algorithm 而非更新的 PACC 或 Randomised Contraction，很大程度上是因为 BigQuery 的 SQL 接口和计费模型与该算法的特性完美匹配。在算法设计中，计算模型和基础设施往往比理论上最优的渐近复杂度更重要。