如何用Python/PySpark/SQL构建细粒度层级依赖映射表

需求概述 #

给定一张包含Source和Target两列的输入表（1000+条记录），需生成层级化的依赖映射表，规则如下：

• 层级定义：L3为最高层级，L0为最低层级
• 链式依赖需完整展开：例如输入A→B、B→C、C→D，输出需呈现为L3=A、L2=B、L1=C、L0=D的结构
• 支持多关联场景：确保每个节点被分配到正确层级

输入表示例：

输出表示例：

实现方案 #

1. 构建依赖关系映射 #

先将输入表的Source-Target关系转换成字典，便于快速查询节点的上下游关联：

import pandas as pd

# 读取输入表（以CSV格式为例）
df = pd.read_csv('input_table.csv')

# 构建下游映射：key为Source，value为对应Target列表
downstream_map = df.groupby('Source')['Target'].apply(list).to_dict()
# 构建上游映射：key为Target，value为对应Source列表
upstream_map = df.groupby('Target')['Source'].apply(list).to_dict()

2. 计算节点层级 #

从无上游依赖的根节点（L3层级）出发，逐层向下遍历，为每个节点标记对应层级：

• L3：无上游关联的根节点
• L2：直接依赖L3的节点
• L1：直接依赖L2的节点
• L0：无下游关联的叶子节点

# 找出所有根节点（不在上游映射中的节点）
all_nodes = set(df['Source'].tolist() + df['Target'].tolist())
root_nodes = [node for node in all_nodes if node not in upstream_map]

# 生成单条层级记录的函数
def build_hierarchy(root):
    hierarchy = {3: [root]}
    current_level = 3
    current_nodes = [root]
    
    while current_level > 0:
        next_level_nodes = []
        for node in current_nodes:
            if node in downstream_map:
                next_level_nodes.extend(downstream_map[node])
        current_level -= 1
        # 填充当前层级节点，无节点则用None占位
        hierarchy[current_level] = next_level_nodes if next_level_nodes else [None]
        current_nodes = next_level_nodes
    
    # 整理为L3到L0的键值对
    return {f'L{level}': nodes[0] for level, nodes in hierarchy.items()}

3. 生成最终输出表 #

遍历所有根节点，生成对应的层级记录，再合并为结构化表格：

output_records = []
for root in root_nodes:
    output_records.append(build_hierarchy(root))

output_df = pd.DataFrame(output_records)
# 按L3到L0的顺序排列列
output_df = output_df[['L3', 'L2', 'L1', 'L0']]
# 保存结果
output_df.to_csv('output_hierarchy.csv', index=False)

多关联场景适配 #

如果存在一个节点对应多个上下游的情况（例如A→B、A→C、B→D、C→D），上述代码会为每个分支生成独立的层级记录，确保每个节点的层级归属准确。示例输入：

对应输出：

（注：若需调整空值显示，可将代码中的None替换为空字符串或其他标记）

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