并查集详解与实现优化

一、什么是并查集？

1. 基本概念

并查集，在一些有N个元素的集合应用问题中，我们通常是在开始时让每个元素构成一个单元素的集合，然后按一定顺序将属于同一组的元素所在的集合合并，其间要反复查找一个元素在哪个集合中。这一类问题近几年来反复出现在信息学的国际国内赛题中，其特点是看似并不复杂，但数据量极大，若用正常的数据结构来描述的话，往往在空间上过大，计算机无法承受；即使在空间上勉强通过，运行的时间复杂度也极高，根本就不可能在比赛规定的运行时间（1～3秒）内计算出试题需要的结果，只能用并查集来描述。

并查集是一种树型的数据结构，用于处理一些不相交集合（Disjoint Sets）的合并及查询问题。常常在使用中以森林来表示。

2. 应用场景

并查集可以高效的解决网络中两点之间的连接问题；
这里说的网络是一个抽象的概念，比如说用户的关系网、地铁站点等都可以叫做网络；

3. 主要操作

并查集支持的主要动作：

• union(p, q)：合并给定的两个参数；
• is_connected(p, q)：查询连个参数是否在同一集合；

4. 并查集接口设计

from abc import ABC, abstractmethod

class UF(ABC):

    @abstractmethod
    def get_size(self):
        pass

    @abstractmethod
    def is_connected(self, p, q):
        pass

    @abstractmethod
    def union_elements(self, p, q):
        pass

二、Quick Find

1. 并查集基本数据表示

对于每个元素并查集存储的是它所属于的集合的id，

从上图中可以发现，0-4这几个数据的id是0，而5-9这几个数据的id是1，我们在这里可以把它理解成两个集合，即0和1；

从这个图示中我们可以得到这样的结论，偶数是一个集合，奇数又是另外一个集合；

从这张图中我们就可以很容易的看出两个点之间是否相连接，比如1和3这两个点肯定是连接的，因为它们在一个集合中，1和2就不可能连接；

上边已经说过对于is_connected(p, q)这个操作实际上是查询两个参数是否在同一集合，我们只需要根据上边图示的概念将它抽象成find(p) == find(q)就可以解决了；

2. 实现第一版并查集

class UnionFind1(UF):

    def __init__(self, size):
        self.id = [i for i in range(size)]

    def get_size(self):
        """获取并查集长度"""
        return len(self.id)

    def is_connected(self, p, q):
        """查看元素p和元素q是否在同一集合"""
        return self.find(p) == self.find(q)

    def union_elements(self, p, q):
        pass

    def find(self, p):
        """查找元素p所对应的集合编号"""
        if p < 0 or p >= len(self.id):
            raise ValueError("p is out of bound.")
        return self.id[p]

3. 实现合并操作

经过我们的union操作以后上图中奇数和偶数所对应的集合id应该是一样的；

其实集合的id取0或者取1都是无所谓的，我们这里取1；

4. 总结

这里我们实现的并查集只是简单的用数组模拟了一下，接下来我们要对现有的并查集进行改造；

三、Quick union

在这里我们将使用树来对并查集进行改造，首先看一下基本的模型：

可以看到这个树结构和之前的树结构不太一样，因为它是由子节点指向父节点的一棵树，

如果我们要合并的话就像图中描述的那样，将需要合并的元素指向一个元素即可；

这样的树结构我们依然可以使用数组来实现：

从这个数组可以看出，我们的10个元素都是父节点，那么我们就可以使用下面的方式进行表述：

此时如果要合并的话，我们只需要改变一下元素的指针即可：

此时的底层数组应该是这样的：

如果此时我们进行union(3, 8)操作的话，就需要这样：

相对应的底层实现应该是这样的：

如果我们此时执行union(9, 4)的操作就会不太一样：

这里我们操作之前需要先执行一次查询的操作；相对应的底层实现就应该是这样的：

通过一系列的union操作以后我们的并查集就变成了这样的：

代码实现(第二版)

class UnionFind2(UF):

    def __init__(self, size):
        self.parent = [i for i in range(size)]

    def get_size(self):
        return len(self.parent)

    def is_connected(self, p, q):
        """查看元素p和元素q是否所在同一集合"""
        return self.find(p) == self.find(q)

    def union_elements(self, p, q):
        """合并"""
        p_root = self.find(p)
        q_root = self.find(q)

        if p_root == q_root:
            return

        self.parent[q_root] = p_root

    def find(self, p):
        """查找p所对应的集合编号"""
        if p < 0 or p >= len(self.parent):
            raise ValueError("p is out of bound.")

        while p != self.parent[p]:
            p = self.parent[p]

        return p

四、基于size的优化

1. 问题分析

如果在极端的情况下，我们的并查集合并操作可能会退化为链表；比如：

执行union(0, 1)：

执行union(0, 2)：

如果一直这样执行下去的话，我们的并查集最终就会变成一个链表结构，这是由于我们在执行合并操作的时候没有对树状的结构进行判断导致的，因为在合并的时候如果没有判断树的形状就会导致树的高度在不断的增加，最终就形成了链表；

2. 解决方案

解决办法就是基于size进行优化：

假设我们现在有这样一个数据结构，然后我们执行union(4, 9)的操作：

假如我们按照之前的思路，肯定会将8执行9这个节点，此时这个树的深度就变成了4，基于size的优化会比较两个树的深度，选择让深度小的树执行深度大的树，最终就变成上图那样了，整个树的最大深度只有3；

基于size的优化实现(第三版)

class UnionFind3(UF):

    def __init__(self, size):
        self.parent = [i for i in range(size)]
        self.sz = [1 for _ in range(size)]

    def get_size(self):
        return len(self.parent)

    def is_connected(self, p, q):
        """查看元素p和元素q是否所在同一集合"""
        return self.find(p) == self.find(q)

    def union_elements(self, p, q):
        """合并"""
        p_root = self.find(p)
        q_root = self.find(q)

        if p_root == q_root:
            return

        # 在合并的时候判断一下合并的集合个数，将个数少的那个集合合并到个数多的那个集合上
        if self.sz[p_root] < self.sz[q_root]:
            self.parent[p_root] = q_root
            self.sz[q_root] += self.sz[p_root]
        else:
            self.parent[q_root] = p_root
            self.sz[p_root] += self.sz[q_root]

    def find(self, p):
        """查找p所对应的集合编号"""
        if p < 0 or p >= len(self.parent):
            raise ValueError("p is out of bound.")

        while p != self.parent[p]:
            p = self.parent[p]

        return p

五、基于rank的优化

1. 问题分析

我们看下边的一个例子：

这也是一个并查集；如果我们才用基于size的优化那样的方式的话，并查集就会变成这样的：

合并后的深度为4；所以更加合理的合并方案是让7这个节点指向8这个节点：

这样合并以后我们得到的新的树的深度并没有发生改变；这样的操作是在合并前将我们先记录一下每一棵树的深度，然后根据深度来进行合并，这样的方式就是基于rank的优化；

基于rank的优化(第四版)

class UnionFind4(UF):

    def __init__(self, size):
        self.parent = [i for i in range(size)]
        # rank[i]表示以i为根的集合所表述的树的层数
        self.rank = [1 for _ in range(size)]

    def get_size(self):
        return len(self.parent)

    def is_connected(self, p, q):
        """查看元素p和元素q是否所在同一集合"""
        return self.find(p) == self.find(q)

    def union_elements(self, p, q):
        """合并"""
        p_root = self.find(p)
        q_root = self.find(q)

        if p_root == q_root:
            return

        # 在合并的时候判断一下合并的集合个数，将个数少的那个集合合并到个数多的那个集合上
        if self.rank[p_root] < self.rank[q_root]:
            self.parent[p_root] = q_root
        elif self.rank[q_root] < self.rank[p_root]:
            self.parent[q_root] = p_root
        else:
            self.parent[q_root] = p_root
            self.rank[p_root] += 1

    def find(self, p):
        """查找p所对应的集合编号"""
        if p < 0 or p >= len(self.parent):
            raise ValueError("p is out of bound.")

        while p != self.parent[p]:
            p = self.parent[p]

        return p

六、路径压缩

1. 问题分析

首先我们需要知道路径压缩解决了什么问题？

这是一个并查集的树；然后在看以下的几个图示：

其实大家已经发现了，这三个图示所表示的概念是一样的；但是由于这三个数所表示的深度不同所以效率肯定是千差万别的；所以我们就需要考虑使用路径压缩来对数的层次进行处理，尽可能的减少树的层次；

2. 路径压缩执行时机

路径压缩最好在什么时候执行？

根据我们的经验，比较好的执行时间是在find操作的时候；如下所示：

当我们在执行查询的时候，我们首先查询的会是最后一层的节点，此时我们执行parent[p] = parent[parent[p]]使我们在执行查询的时候只关注当前节点的根节点即可；

最终的结果就会变成这样的；此时当尝试继续往下执行的时候，会发现已经到达根节点，所以就不会执行了，此时这棵树的深度就由5变成了3；

路径压缩实现(第五版)

class UnionFind5(UF):

    def __init__(self, size):
        self.parent = [i for i in range(size)]
        # rank[i]表示以i为根的集合所表述的树的层数
        self.rank = [1 for _ in range(size)]

    def get_size(self):
        return len(self.parent)

    def is_connected(self, p, q):
        """查看元素p和元素q是否所在同一集合"""
        return self.find(p) == self.find(q)

    def union_elements(self, p, q):
        """合并"""
        p_root = self.find(p)
        q_root = self.find(q)

        if p_root == q_root:
            return

        # 在合并的时候判断一下合并的集合个数，将个数少的那个集合合并到个数多的那个集合上
        if self.rank[p_root] < self.rank[q_root]:
            self.parent[p_root] = q_root
        elif self.rank[q_root] < self.rank[p_root]:
            self.parent[q_root] = p_root
        else:
            self.parent[q_root] = p_root
            self.rank[p_root] += 1

    def find(self, p):
        """查找p所对应的集合编号"""
        if p < 0 or p >= len(self.parent):
            raise ValueError("p is out of bound.")

        while p != self.parent[p]:
            # 路径压缩
            self.parent[p] = self.parent[self.parent[p]]
            p = self.parent[p]

        return p

七、递归实现路径压缩(第六版)

class UnionFind6(UF):

    def __init__(self, size):
        self.parent = [i for i in range(size)]
        # rank[i]表示以i为根的集合所表述的树的层数
        self.rank = [1 for _ in range(size)]

    def get_size(self):
        return len(self.parent)

    def is_connected(self, p, q):
        """查看元素p和元素q是否所在同一集合"""
        return self.find(p) == self.find(q)

    def union_elements(self, p, q):
        """合并"""
        p_root = self.find(p)
        q_root = self.find(q)

        if p_root == q_root:
            return

        # 在合并的时候判断一下合并的集合个数，将个数少的那个集合合并到个数多的那个集合上
        if self.rank[p_root] < self.rank[q_root]:
            self.parent[p_root] = q_root
        elif self.rank[q_root] < self.rank[p_root]:
            self.parent[q_root] = p_root
        else:
            self.parent[q_root] = p_root
            self.rank[p_root] += 1

    def find(self, p):
        """查找p所对应的集合编号"""
        if p < 0 or p >= len(self.parent):
            raise ValueError("p is out of bound.")

        if p != self.parent[p]:
            # 路径压缩
            self.parent[p] = self.find(self.parent[p])

        return self.parent[p]

八、六种版本并查集的性能测试对比

1. 测试代码

import random
import time

def test_uf(uf, m):
    size = uf.get_size()

    start_time = time.time()

    for i in range(m):
        a = random.randint(0, size - 1)
        b = random.randint(0, size - 1)
        uf.union_elements(a, b)

    for i in range(m):
        a = random.randint(0, size - 1)
        b = random.randint(0, size - 1)
        uf.is_connected(a, b)

    end_time = time.time()

    elapsed_time = end_time - start_time
    return elapsed_time

if __name__ == "__main__":
    size = 100000
    m = 100000

    uf1 = UnionFind1(size)
    print(f"UnionFind1 : {test_uf(uf1, m)} s")

    uf2 = UnionFind2(size)
    print(f"UnionFind2 : {test_uf(uf2, m)} s")

    uf3 = UnionFind3(size)
    print(f"UnionFind3 : {test_uf(uf3, m)} s")

    uf4 = UnionFind4(size)
    print(f"UnionFind4 : {test_uf(uf4, m)} s")

    uf5 = UnionFind5(size)
    print(f"UnionFind5 : {test_uf(uf5, m)} s")

    uf6 = UnionFind6(size)
    print(f"UnionFind6 : {test_uf(uf6, m)} s")

总结

通过以上六个版本的并查集实现和性能测试，我们可以看到：

1. Quick Find (第一版)：查询效率高O(1)，但合并操作效率低O(n)
2. Quick Union (第二版)：改善了合并操作，但在极端情况下可能退化为链表
3. 基于Size优化 (第三版)：通过维护集合大小，避免树过深的问题
4. 基于Rank优化 (第四版)：通过维护树的深度，进一步优化合并策略
5. 路径压缩优化 (第五版)：在查找过程中压缩路径，大幅提升后续操作效率
6. 递归路径压缩 (第六版)：使用递归实现完全的路径压缩

从性能测试结果可以看出，随着优化的深入，并查集的性能得到了显著提升，特别是在大数据量的情况下，优化后的版本表现出色。