求SQL/Python实现多父多子场景下家族树成员分组方案
多对多父-子关联关系的家族分组实现方案
问题背景
现有一张存储父-子关联关系的表,支持一个父节点对应多个子节点、一个子节点对应多个父节点,无关联项不在表中。示例表结构及数据如下:
| 父节点 | 子节点 |
|---|---|
| A | B |
| A | C |
| A | D |
| D | E |
| G | H |
| I | J |
| J | K |
| L | M |
| O | M |
| N | P |
| Q | R |
| R | P |
期望输出为按家族分组的成员列表,示例如下:
Family 1 = [A, B, C, D, E] Family 2 = [G, H] Family 3 = [I, J, K] Family 4 = [L, M, O] Family 5 = [P, N, Q, R]
已尝试两种方案但存在问题:
- 使用SQL递归查询查找顶级祖先再遍历后代,但因顶级祖先可能有多个,难以处理跨关联的合并场景;
- 使用Python逐行循环分组,遇到跨家族关联时需合并分组,逻辑复杂且效率低。
现有约15000条关联数据,需高效实现分组。
一、Python 高效实现方案(并查集/Union-Find)
处理这类连通分量问题,并查集是最优选择,时间复杂度接近O(n),适合大规模数据。
实现代码
class UnionFind: def __init__(self): self.parent = {} 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): # 初始化节点 if x not in self.parent: self.parent[x] = x if y not in self.parent: self.parent[y] = y # 合并两个集合 root_x = self.find(x) root_y = self.find(y) if root_x != root_y: self.parent[root_y] = root_x # 模拟从数据库读取的关联数据 relations = [ ("A", "B"), ("A", "C"), ("A", "D"), ("D", "E"), ("G", "H"), ("I", "J"), ("J", "K"), ("L", "M"), ("O", "M"), ("N", "P"), ("Q", "R"), ("R", "P") ] # 初始化并查集 uf = UnionFind() for parent_node, child_node in relations: uf.union(parent_node, child_node) # 按根节点分组 families = {} for node in uf.parent: root = uf.find(node) if root not in families: families[root] = [] families[root].append(node) # 输出结果,按家族编号排序 for idx, (root, members) in enumerate(sorted(families.values(), key=lambda x: x[0]), 1): print(f"Family {idx} = [{', '.join(sorted(members))}]")
说明
- 并查集通过
find(路径压缩)和union(合并集合)操作,高效维护节点的连通关系; - 遍历所有关联对,将父节点和子节点合并到同一集合;
- 最后按根节点分组,得到每个家族的所有成员;
- 15000条数据可在毫秒级完成处理。
二、SQL 实现方案(递归CTE + 分组)
SQL处理这类问题需要通过递归收集所有连通节点,再进行分组,适合直接在数据库中处理的场景。
实现代码(以MySQL 8.0+为例)
-- 1. 创建临时表存储所有关联关系(如果已有表可跳过) CREATE TEMPORARY TABLE IF NOT EXISTS parent_child ( parent_node VARCHAR(10), child_node VARCHAR(10) ); INSERT INTO parent_child VALUES ('A','B'),('A','C'),('A','D'),('D','E'), ('G','H'),('I','J'),('J','K'),('L','M'), ('O','M'),('N','P'),('Q','R'),('R','P'); -- 2. 递归CTE收集所有连通节点 WITH RECURSIVE family_relations AS ( -- 初始:所有节点作为自身的连通节点 SELECT parent_node AS node, parent_node AS root_node FROM parent_child UNION SELECT child_node AS node, parent_node AS root_node FROM parent_child UNION -- 递归:查找所有关联的节点,传递根节点 SELECT fr.node, pc.parent_node AS root_node FROM family_relations fr JOIN parent_child pc ON fr.root_node = pc.child_node UNION SELECT fr.node, pc.child_node AS root_node FROM family_relations fr JOIN parent_child pc ON fr.root_node = pc.parent_node ), -- 3. 去重并找到每个节点的最小根节点(用于分组) unique_nodes AS ( SELECT node, MIN(root_node) AS family_root FROM family_relations GROUP BY node ) -- 4. 按家族根节点分组,输出结果 SELECT CONCAT('Family ', ROW_NUMBER() OVER(ORDER BY family_root), ' = [', GROUP_CONCAT(DISTINCT node ORDER BY node SEPARATOR ', '), ']') AS family_result FROM unique_nodes GROUP BY family_root ORDER BY family_root;
说明
- 递归CTE
family_relations会遍历所有节点的双向关联,收集每个节点能连通的所有根节点; unique_nodes通过取最小根节点来统一分组标识;- 最后用
GROUP_CONCAT合并同组节点,生成期望格式的输出; - 对于15000条数据,需确保数据库配置足够(如递归深度、临时表内存),效率略低于Python但可满足需求。
内容的提问来源于stack exchange,提问作者Pretzel Stands
相关产品推荐
相关产品推荐

