求最优算法：将关联成员归入同一Bucket（分组合并问题）

问题描述 #

给定群组列表（示例如下），需将成员划分到Bucket中，遵循以下规则：

• 若成员无对应Bucket则新建
• 若当前群组有成员已在某Bucket中，整个群组归入该Bucket
• 若群组成员分属多个Bucket，则合并这些Bucket

示例输入：

groups = [
    ['A', 'B', 'C'],
    ['D', 'A', 'E'],
    ['X', 'Y', 'Z'],
    ['X', 'A', 'W'],
    ['AA', 'BB', 'CC'],
]

示例最终应得到2个Bucket：{'A','B','C','D','E','X','Y','Z','W'} 和 {'AA','BB','CC'}

现针对数千群组、每组数百成员的场景，提供高效实现方案。

核心实现思路：并查集（Union-Find） #

并查集是专门处理动态连通性问题的数据结构，它的查找（Find）和合并（Union）操作均摊时间复杂度接近O(1)，完全适配大规模群组的合并需求。

具体步骤 #

1. 初始化映射表：用字典parent记录每个成员的父节点，初始时每个成员的父节点是自身（代表独立Bucket）；用rank字典记录每个Bucket的层级高度，用于优化合并效率。
2. 遍历所有群组：
   • 先将所有成员加入并查集，确保每个成员都有初始的父节点和秩。
   • 对每个群组，取第一个成员作为基准，将群组内其他成员依次与基准执行合并操作：若成员分属不同Bucket，则合并这两个Bucket。
3. 整理最终Bucket：遍历所有成员，通过查找操作定位每个成员所属的根节点，将同一根节点下的成员归为同一个Bucket。

关键优化 #

• 路径压缩：查找成员根节点时，直接将成员指向根节点，减少后续查找的层级，大幅提升重复查找效率。
• 按秩合并：合并时将层级较低的Bucket挂载到层级较高的Bucket根节点下，避免树结构退化为链表，保证操作的时间效率。

代码实现（Python） #

class UnionFind:
    def __init__(self):
        self.parent = {}
        self.rank = {}

    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):
        x_root = self.find(x)
        y_root = self.find(y)
        if x_root == y_root:
            return
        # 按秩合并优化
        if self.rank[x_root] < self.rank[y_root]:
            self.parent[x_root] = y_root
        else:
            self.parent[y_root] = x_root
            if self.rank[x_root] == self.rank[y_root]:
                self.rank[x_root] += 1

    def add_member(self, x):
        if x not in self.parent:
            self.parent[x] = x
            self.rank[x] = 0

def group_into_buckets(groups):
    uf = UnionFind()
    # 先将所有成员注册到并查集
    for group in groups:
        for member in group:
            uf.add_member(member)
    # 处理每个群组的合并逻辑
    for group in groups:
        if not group:
            continue
        base_member = group[0]
        for member in group[1:]:
            uf.union(base_member, member)
    # 整理生成最终的Bucket集合
    buckets = {}
    for member in uf.parent:
        root = uf.find(member)
        if root not in buckets:
            buckets[root] = set()
        buckets[root].add(member)
    return list(buckets.values())

# 测试示例
groups = [
    ['A', 'B', 'C'],
    ['D', 'A', 'E'],
    ['X', 'Y', 'Z'],
    ['X', 'A', 'W'],
    ['AA', 'BB', 'CC'],
]

print(group_into_buckets(groups))
# 输出：[{'A', 'B', 'C', 'D', 'E', 'X', 'Y', 'Z', 'W'}, {'AA', 'BB', 'CC'}]

性能说明 #

• 初始化阶段：遍历所有成员，时间复杂度O(N)，N为总成员数。
• 合并阶段：每个Union操作均摊时间复杂度为O(α(N))，α是阿克曼函数的反函数，增长极慢，百万级数据量下几乎等同于常数时间。
• 整理Bucket阶段：遍历所有成员，时间复杂度O(N)。

针对数千群组、每组数百成员的场景（总成员数可达数百万），该方案的效率远高于暴力查找合并的方法，能稳定快速完成划分任务。

内容的提问来源于stack exchange，提问作者Chris