You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

求SQL/Python实现多父多子场景下家族树成员分组方案

多对多父-子关联关系的家族分组实现方案

问题背景

现有一张存储父-子关联关系的表,支持一个父节点对应多个子节点、一个子节点对应多个父节点,无关联项不在表中。示例表结构及数据如下:

父节点子节点
AB
AC
AD
DE
GH
IJ
JK
LM
OM
NP
QR
RP

期望输出为按家族分组的成员列表,示例如下:

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))}]")

说明

  1. 并查集通过find(路径压缩)和union(合并集合)操作,高效维护节点的连通关系;
  2. 遍历所有关联对,将父节点和子节点合并到同一集合;
  3. 最后按根节点分组,得到每个家族的所有成员;
  4. 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;

说明

  1. 递归CTEfamily_relations会遍历所有节点的双向关联,收集每个节点能连通的所有根节点;
  2. unique_nodes通过取最小根节点来统一分组标识;
  3. 最后用GROUP_CONCAT合并同组节点,生成期望格式的输出;
  4. 对于15000条数据,需确保数据库配置足够(如递归深度、临时表内存),效率略低于Python但可满足需求。

内容的提问来源于stack exchange,提问作者Pretzel Stands

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.06 21:23:24