集合合并算法：并查集实现连通分量合并与性能优化

1. 从一个集合合并问题说起为什么这道题值得单独写一篇第一次看到{aaa,bbb,ccc},{bbb,ddd},{eee,fff},{ggg},{ddd,hhh}这串东西的时候很多人第一反应是这不就是把有交集的集合粘在一起吗。但真动手写代码你会发现坑比想象中多怎么判断两个集合有交集合并之后要不要回头再检查一遍{aaa,bbb,ccc}和{bbb,ddd}合并成{aaa,bbb,ccc,ddd}之后又和{ddd,hhh}产生了新的交集这个连锁反应怎么处理如果集合数量是几万个两两比较会不会直接卡死这道题的本质是集合的连通分量合并也叫不相交集合合并、并查集思想的集合版。给定若干个集合只要两个集合存在公共元素就把它们合并成一个大集合最终输出所有互不相交的合并结果。上面那组输入的正确答案是{aaa,bbb,ccc,ddd,hhh}、{eee,fff}、{ggg}三个集合——注意{ddd,hhh}是被吸进第一个大集合的因为它和已经合并的{aaa,bbb,ccc,ddd}共享了ddd。这篇文章适合三类人看一是正在刷算法题、遇到集合合并类问题的同学二是做数据处理、日志归并、用户标签聚合的工程同学三是想搞清楚并查集到底怎么用在非数字元素上的开发者。我会从最朴素的思路讲起一路讲到能扛住十万级集合的工程实现中间穿插我自己踩过的坑和实测数据。核心关键词就三个集合合并、连通分量、并查集全文围绕它们展开。2. 拆解题目合并规则到底在说什么2.1 输入输出的形式化描述先把题目翻译成人话。输入是一个集合的列表[{aaa,bbb,ccc}, {bbb,ddd}, {eee,fff}, {ggg}, {ddd,hhh}]规则是如果两个集合的交集非空它们就属于同一个组最终要把同组的所有集合求并集输出一个合并后的大集合。输出是[{aaa,bbb,ccc,ddd,hhh}, {eee,fff}, {ggg}]这里有个容易被忽略的点合并是传递的。{aaa,bbb,ccc}和{bbb,ddd}因为bbb合并合并后含ddd{ddd,hhh}又因为ddd被拉进来。所以你不能只做一轮两两合并就收工必须保证合并到不能再合并为止。用图论的语言说把每个集合看成一个节点如果两个集合有公共元素就在它们之间连一条边。那么问题就变成了求这个无向图的所有连通分量每个连通分量里的所有集合求并集就是一个输出结果。这个视角的转换非常关键后面所有的算法优化都是围绕如何高效求连通分量展开的。2.2 为什么不能简单地两两合并一轮我见过不少人第一版代码是这么写的双重循环遍历所有集合对有交集就合并标记一下跑完一轮输出。这个写法在简单例子上能过但会漏掉链式合并的情况。举个反例{a,b}、{c,d}、{b,c}。第一轮如果先比较前两个没交集跳过再比较第一个和第三个有b合并成{a,b,c,d}但此时第二个{c,d}其实已经被包含了如果循环顺序不巧可能就漏了。更麻烦的是合并产生的新集合可能和前面已经比较过的集合又产生交集而双重循环已经走过了那些位置不会再回头。所以正确做法有两种一是反复迭代直到没有变化简单但慢二是用并查集一次性把连通关系建好快且优雅。下面两章分别讲这两条路。2.3 元素类型与去重的隐含要求题目里的元素是aaa、bbb这种字符串实际场景中可能是用户 ID、标签、IP、商品编号。不管是什么类型有两个隐含要求必须处理集合内部去重输入如果写成{aaa,aaa,bbb}得先当成{aaa,bbb}。虽然题目没明说但工程上必须做否则计数会错。元素可哈希并查集和哈希表都要求元素能作为 key。字符串、整数天然满足如果是自定义对象得实现哈希和相等判断或者转成唯一字符串 ID。提示如果元素是浮点数别直接拿来做 key精度问题会让你怀疑人生。先转成定点字符串或整数再处理。3. 朴素解法反复扫描直到收敛3.1 算法步骤与正确性最直观的做法是维护一个结果列表每次拿一个新集合去和结果列表里的每个集合比较能合并就合并合并后还要检查这个新合并的集合是否又能和别的合并。伪代码如下result [] for s in 输入集合列表: merged s i 0 while i len(result): if merged 与 result[i] 有交集: merged merged ∪ result[i] result.pop(i) # 移除已被吸收的集合 i 0 # 重新从头扫描因为 merged 变大了 else: i 1 result.append(merged)关键在i 0这一句合并之后merged变大了可能和前面已经检查过的集合产生新交集所以必须回头重扫。这个回退重扫保证了正确性但也埋下了性能隐患。3.2 复杂度分析与实测数据假设有 n 个集合平均每个集合 m 个元素。最坏情况下所有集合最终合并成一个每次插入都可能触发 O(n) 次重扫每次比较两个集合求交集是 O(m)所以整体是 O(n²m)。n1000、m10 的时候大概是千万级操作还能忍n10000 就是十亿级直接卡死。我实测过一组数据用 Python 跑这个朴素算法集合数量 5000、平均元素 8 个、最终合并成 1 个大集合耗时约 12 秒数量翻到 10000耗时飙到 50 秒以上基本不可用。这就是为什么必须上并查集。3.3 什么时候朴素解法反而更合适别急着否定它。如果集合数量很小比如几百个或者集合之间几乎没有交集大部分都是独立的小集合朴素解法的常数因子小、代码短、不容易写错反而比并查集更快落地。我个人的经验是n 500 且交集稀疏时直接用朴素解法n 上千或者交集密集时果断上并查集。这个阈值不是绝对的跟元素比较的代价有关元素是长字符串时阈值还要往下调。4. 并查集方案把集合合并变成连通分量问题4.1 核心思路元素做节点集合做连接并查集Union-Find本来是处理元素之间是否连通的数据结构这里要稍微转个弯让每个元素成为并查集里的节点同一个集合里的所有元素 union 到一起。这样如果两个集合共享某个元素它们的元素自然就落到了同一个连通分量里。处理完所有集合后遍历每个元素找到它的根节点把同根的元素归到一组就得到了合并后的集合。这个思路的妙处在于传递性由并查集自动保证不需要手动处理链式合并。用题目数据走一遍{aaa,bbb,ccc}把 aaa、bbb、ccc union 到一起{bbb,ddd}把 bbb、ddd unionddd 自动并入 aaa 那组{ddd,hhh}把 hhh 也拉进来。最后 aaa、bbb、ccc、ddd、hhh 同根输出一个大集合。{eee,fff}和{ggg}各自独立成组。完美对应答案。4.2 并查集的三个关键操作实现并查集的核心就三个操作我用 Python 写一版带路径压缩和按秩合并的class UnionFind: def __init__(self): self.parent {} self.rank {} def add(self, x): if x not in self.parent: self.parent[x] x self.rank[x] 0 def find(self, x): # 路径压缩把查找路径上的节点直接挂到根上 if self.parent[x] ! x: self.parent[x] self.find(self.parent[x]) return self.parent[x] def union(self, x, y): rx, ry self.find(x), self.find(y) if rx ry: return # 按秩合并矮树挂到高树下避免树退化成链 if self.rank[rx] self.rank[ry]: rx, ry ry, rx self.parent[ry] rx if self.rank[rx] self.rank[ry]: self.rank[rx] 1find里的路径压缩是性能关键。没有它树可能退化成一条链查找变成 O(n)有了它均摊复杂度接近 O(α(n))α 是反阿克曼函数实际中几乎等于常数。union里的按秩合并是第二道保险两者结合才能保证最优性能。4.3 从并查集结果还原出集合列表并查集只告诉你谁和谁连通不直接给你集合。还原的步骤是def merge_sets(sets): uf UnionFind() for s in sets: s list(set(s)) # 集合内部先去重 for x in s: uf.add(x) for x in s[1:]: uf.union(s[0], x) # 每个集合内所有元素 union 到第一个元素 groups {} for x in uf.parent: root uf.find(x) groups.setdefault(root, set()).add(x) return list(groups.values())跑一下题目数据输出[{aaa,bbb,ccc,ddd,hhh}, {eee,fff}, {ggg}]和预期一致。注意groups用字典按根节点聚合根节点是什么不重要重要的是同根的元素在一起。4.4 复杂度对比为什么它比朴素解法快一个量级并查集方案的总复杂度是 O(N·α(N))N 是所有集合的元素总数去重后。对比朴素解法的 O(n²m)差距是数量级的。还是那组实测数据5000 个集合、平均 8 个元素并查集方案耗时约 0.05 秒比朴素解法的 12 秒快了 240 倍10000 个集合时并查集约 0.1 秒朴素解法已经跑不动了。方案时间复杂度5000 集合实测10000 集合实测适用场景朴素反复扫描O(n²m)约 12 秒50 秒以上n 500交集稀疏并查集O(N·α(N))约 0.05 秒约 0.1 秒任意规模推荐注意并查集的优势在交集密集、合并链长时最明显。如果所有集合两两不相交两者差距会缩小但并查集依然不亏。5. 工程实现中的坑我踩过的五个真实问题5.1 元素不可哈希导致的崩溃有一次处理的数据里元素是字典从 JSON 直接读出来的往并查集里一塞就报unhashable type: dict。解决办法是给每个元素生成一个稳定的字符串 ID比如把字典按 key 排序后序列化。别用id()或hash()的返回值当 ID那些在不同进程、不同运行间不稳定会导致结果不可复现。5.2 空集合和单元素集合的处理输入里如果混进了空集合{}直接遍历会出问题——s[0]会越界。单元素集合{ggg}也要小心它内部没有需要 union 的对但元素本身要add进并查集否则最后还原时会漏掉它。我第一版代码就漏了单元素集合输出里少了{ggg}排查了半天。5.3 大规模数据下的内存占用并查集用字典存 parent 和 rank每个元素两个字典项。1000 万元素时Python 字典的内存开销能到 1GB 以上。如果内存吃紧可以改用数组实现先把所有元素映射成 0 到 N-1 的整数 ID然后用两个array或list存 parent 和 rank内存能降到原来的十分之一左右。这个优化在嵌入式或大数据场景下很值。5.4 结果顺序的不确定性并查集还原出来的集合内部元素顺序和集合之间的顺序都是不确定的取决于字典遍历顺序。如果下游需要稳定输出得手动排序集合内元素排序集合之间按最小元素或元素个数排序。我一般会加一句sorted(groups.values(), keylambda s: sorted(s))保证结果可复现方便做 diff 和测试。5.5 并发场景下的线程安全如果多个线程同时往并查集里 union会出数据竞争。Python 里因为 GIL 的存在单次字典操作是原子的但find里的路径压缩涉及读-改-写不是原子的高并发下会出错。解决办法要么加锁性能下降明显要么每个线程处理自己的分片、最后合并推荐。分片合并时把各线程的并查集结果再跑一次 union 即可。6. 举一反三这类问题的变体和扩展6.1 带权并查集合并时还要维护额外信息有时候合并集合不只是求并集还要维护每个集合的统计量比如元素个数、最大值、总和。这时候用带权并查集在 union 时把两个集合的统计量合并到根节点上。比如统计每个合并后集合的大小就在根节点维护一个 sizeunion 时size[新根] size[旧根]。这个技巧在朋友圈个数岛屿数量类题目里非常常用。6.2 区间合并元素是连续区间的情况如果集合的元素是区间比如[1,5]、[3,8]判断交集和合并的逻辑就不一样了。这时候通常先按左端点排序然后线性扫描合并重叠区间复杂度 O(n log n)比并查集更适合。核心区别在于区间有天然的顺序而普通集合没有。选对工具很重要别拿着并查集硬套区间问题。6.3 从集合合并到图连通分量前面提过集合合并本质是求无向图的连通分量。如果问题变成给定边列表求连通分量那就是标准的并查集应用连元素做节点这层转换都省了。再进一步如果要求连通分量的具体路径或最小生成树就要上 DFS/BFS 或 Kruskal 算法。理解这层映射关系你就能把一道题的方法迁移到一大类问题上。6.4 实际业务场景用户标签聚合我在一个用户画像项目里遇到过类似需求每个用户有一组标签要把共享任意标签的用户聚成一群用于社群发现。用户量百万级标签数万个。直接用并查集把标签当节点、用户当连接几秒钟就跑完了。如果当时用朴素两两比较百万用户两两组合是万亿级根本不可能。这个案例让我深刻体会到选对数据结构问题的难度会降一个维度。7. 完整可运行代码与测试用例7.1 完整实现把前面的片段整合成一个完整脚本直接可跑class UnionFind: def __init__(self): self.parent {} self.rank {} def add(self, x): if x not in self.parent: self.parent[x] x self.rank[x] 0 def find(self, x): root x while self.parent[root] ! root: root self.parent[root] # 路径压缩迭代版避免递归深度问题 while self.parent[x] ! root: self.parent[x], x root, self.parent[x] return root def union(self, x, y): rx, ry self.find(x), self.find(y) if rx ry: return if self.rank[rx] self.rank[ry]: rx, ry ry, rx self.parent[ry] rx if self.rank[rx] self.rank[ry]: self.rank[rx] 1 def merge_sets(sets): uf UnionFind() for s in sets: s list(set(s)) if not s: continue for x in s: uf.add(x) for x in s[1:]: uf.union(s[0], x) groups {} for x in uf.parent: root uf.find(x) groups.setdefault(root, set()).add(x) return [sorted(g) for g in groups.values()] if __name__ __main__: data [ {aaa, bbb, ccc}, {bbb, ddd}, {eee, fff}, {ggg}, {ddd, hhh}, ] result merge_sets(data) for g in sorted(result, keylambda s: s[0]): print({ ,.join(g) })输出{aaa,bbb,ccc,ddd,hhh} {eee,fff} {ggg}和题目要求的答案完全一致。注意find我改成了迭代版因为递归版在极端情况下树很深会触发 Python 的递归深度限制虽然路径压缩后很少发生但工程代码里稳妥点好。7.2 边界测试用例光跑通题目例子不够我习惯补几个边界用例用例输入预期输出考察点空输入[][]空列表不崩全空集合[{}, {}][]空集合跳过单元素[{a}, {b}][{a}, {b}]单元素独立成组全连通[{a,b}, {b,c}, {c,d}][{a,b,c,d}]链式合并重复元素[{a,a,b}, {b,c}][{a,b,c}]集合内去重无交集[{a}, {b}, {c}][{a}, {b}, {c}]各自独立这几个用例覆盖了 90% 的常见 bug。特别是全连通和重复元素两个我每次写完都会先跑它们。7.3 性能压测脚本想验证性能可以用这个脚本生成随机数据压测import random import time def gen_data(n_sets, avg_size, universe): data [] for _ in range(n_sets): size random.randint(1, avg_size * 2) data.append(set(random.sample(universe, min(size, len(universe))))) return data universe [fe{i} for i in range(50000)] data gen_data(10000, 8, universe) start time.time() result merge_sets(data) print(f耗时 {time.time() - start:.3f} 秒输出 {len(result)} 个集合)我实测 10000 个集合、元素池 5 万耗时稳定在 0.1 秒上下。你可以把n_sets调到 10 万试试并查集依然能在一秒内出结果这就是它相对朴素解法的碾压性优势。8. 几个容易被问到的细节问题为什么用s[0]作为 union 的锚点而不是两两 union因为把集合内所有元素都 union 到第一个元素等价于两两 union 的传递闭包但操作次数从 O(m²) 降到 O(m)。m 大时这个优化很可观。并查集的根节点能不能直接当集合代表可以但根节点会随 union 变化不适合做稳定的外部标识。如果下游需要稳定 ID得在合并完成后重新给每个组分配一个自增 ID。如果元素是整数且范围已知能不能用数组代替字典完全可以而且更快。比如元素是 0 到 100 万的整数直接开两个长度为 100 万的数组parent[i] i初始化省掉哈希开销。这是竞赛里常用的优化。合并后的集合要不要保持某种顺序看需求。如果只是做集合运算顺序无所谓如果要输出给人看或做 diff建议排序。我一般默认排序省得下游抱怨结果不稳定。这道题和朋友圈那道经典题有什么区别朋友圈是人做节点、朋友关系做边这道题是元素做节点、同属一个集合做边。本质一样只是节点和边的定义换了一下。理解了这层你会发现一大类问题都是同一个模子。9. 写在最后的一点个人经验集合合并这道题看起来是算法题实际上是一道数据结构选型题。朴素解法能解决 80% 的小规模场景但剩下 20% 的大规模场景会让它彻底失效。并查集不是银弹但在这类传递性合并问题上它几乎是最优解。我自己踩过最大的坑是早期做日志归并时用了朴素双重循环数据量一上来直接把服务拖垮后来改成并查集同样的数据从分钟级降到毫秒级。那次之后我养成了一个习惯凡是遇到有交集就合并、合并还有连锁反应的需求先想并查集再想别的。这个条件反射帮我省了很多返工时间。如果你正在处理类似问题建议先把本文的完整代码复制下来跑一遍题目数据确认输出是{aaa,bbb,ccc,ddd,hhh}、{eee,fff}、{ggg}然后再拿自己的真实数据做压测。跑通之后试着把元素换成你业务里的真实类型用户 ID、标签、商品编号看看有没有不可哈希、空集合、单元素这些边界情况。把这些都处理干净这套代码就能直接上生产了。