基于id与sibling_id生成family_id列的技术实现需求

用户需要将DataFrame中存在亲属关系（含互为兄弟姐妹）的id归为同一family_id，无亲属关系的id单独分配家庭编号。以下是具体实现方案：

原始数据 #

import pandas as pd
import numpy as np

df = pd.DataFrame({
    'id': [1, 1, 2, 2, 3, 3, 4, 5, 6, 7],
    'field_a': list('AABBCCDEFG'),
    'sibling_id': [2, 3, 1, 3, 1, 2, np.nan, np.nan, 7, 6],
    'sibling_field_a': ['B', 'C', 'A', 'C' , 'A', 'B', np.nan, np.nan, 'G', 'F']
})

df['sibling_id'] = df['sibling_id'].astype('Int64')

方法1：使用NetworkX处理连通分量 #

利用NetworkX的图结构识别连通分量，快速完成家庭分组：

import networkx as nx

# 提取有效亲属关系边
edges = df.dropna(subset=['sibling_id'])[['id', 'sibling_id']].values.tolist()

# 构建无向图并添加边
G = nx.Graph()
G.add_edges_from(edges)

# 为每个连通分量分配family_id
family_mapping = {}
for family_id, component in enumerate(nx.connected_components(G)):
    for node in component:
        family_mapping[node] = family_id

# 为无亲属关系的id分配新编号
max_family_id = max(family_mapping.values()) if family_mapping else -1
for idx in df['id'].unique():
    if idx not in family_mapping:
        max_family_id += 1
        family_mapping[idx] = max_family_id

# 映射到原DataFrame
df['family_id'] = df['id'].map(family_mapping)

方法2：并查集（Union-Find）算法（无第三方库依赖） #

手动实现并查集逻辑，无需额外安装库：

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
        x_root = self.find(x)
        y_root = self.find(y)
        if x_root != y_root:
            self.parent[y_root] = x_root

# 初始化并查集并合并亲属关系
uf = UnionFind()
for _, row in df.dropna(subset=['sibling_id']).iterrows():
    uf.union(row['id'], row['sibling_id'])

# 为连通分量分配family_id
family_mapping = {}
current_id = 0
roots = set()

# 处理有亲属关系的节点
for node in uf.parent:
    root = uf.find(node)
    if root not in roots:
        roots.add(root)
        family_mapping[root] = current_id
        current_id += 1
for node in uf.parent:
    family_mapping[node] = family_mapping[uf.find(node)]

# 处理无亲属关系的节点
for idx in df['id'].unique():
    if idx not in family_mapping:
        family_mapping[idx] = current_id
        current_id += 1

# 添加family_id列
df['family_id'] = df['id'].map(family_mapping)

两种方法均可生成符合需求的family_id列，运行后得到的结果与期望输出一致。

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