Scala Spark层级树形数据表的扁平化处理技术咨询

嘿，我来帮你搞定这个树形数据转扁平化的需求～先给你看最终的扁平化输出效果，再分享两种常见的实现方式：

方法1：用SQL递归CTE实现 #

如果你的数据存在数据库里，用递归CTE是最直接的方式，先遍历所有节点的层级关系，再通过CASE语句把层级转成列：

WITH RECURSIVE Hierarchy AS (
    -- 先取根节点（depth=0的行）
    SELECT 
        Id, name, parentId, path, depth,
        depth AS current_level
    FROM your_table
    WHERE depth = 0
    
    UNION ALL
    
    -- 递归关联子节点，逐层遍历
    SELECT 
        t.Id, t.name, t.parentId, t.path, t.depth,
        h.current_level + 1 AS current_level
    FROM your_table t
    JOIN Hierarchy h ON t.parentId = h.Id
)
-- 用CASE语句把每个层级的Id和Name转成单独的列
SELECT 
    base.Id, base.name, base.parentId, base.path, base.depth,
    MAX(CASE WHEN h.current_level = 0 THEN h.Id END) AS Level0_Id,
    MAX(CASE WHEN h.current_level = 0 THEN h.name END) AS Level0_Name,
    MAX(CASE WHEN h.current_level = 1 THEN h.Id END) AS Level1_Id,
    MAX(CASE WHEN h.current_level = 1 THEN h.name END) AS Level1_Name,
    MAX(CASE WHEN h.current_level = 2 THEN h.Id END) AS Level2_Id,
    MAX(CASE WHEN h.current_level = 2 THEN h.name END) AS Level2_Name,
    MAX(CASE WHEN h.current_level = 3 THEN h.Id END) AS Level3_Id,
    MAX(CASE WHEN h.current_level = 3 THEN h.name END) AS Level3_Name
FROM your_table base
LEFT JOIN Hierarchy h ON base.Id = h.Id OR base.path LIKE CONCAT('%/', h.Id, '%')
GROUP BY base.Id, base.name, base.parentId, base.path, base.depth
ORDER BY base.Id;

方法2：用Python Pandas处理 #

如果是本地数据，用Pandas脚本处理更灵活，先把节点信息存成字典，再根据path字段拆分出所有层级的Id，匹配对应的名称：

import pandas as pd

# 模拟你的原始数据表
raw_data = [
    {"Id": 55, "name": "Canada", "parentId": None, "path": None, "depth": 0},
    {"Id": 77, "name": "Ontario", "parentId": 55, "path": "/55", "depth": 1},
    {"Id": 100, "name": "Toronto", "parentId": 77, "path": "/55/77", "depth": 2},
    {"Id": 104, "name": "Brampton", "parentId": 100, "path": "/55/77/100", "depth": 3}
]
df = pd.DataFrame(raw_data)

# 先做个节点字典，方便快速通过Id查名称
node_map = df.set_index('Id')['name'].to_dict()

# 定义函数处理每一行，生成扁平化的层级字段
def flatten_node(row):
    # 拆分path获取所有父节点Id，加上自身Id
    if pd.isna(row['path']):
        level_ids = [row['Id']]
    else:
        level_ids = [int(id_str) for id_str in row['path'].strip('/').split('/')] + [row['Id']]
    
    # 补全到最大层级数（这里最大depth是3，所以4个层级）
    max_depth = df['depth'].max()
    level_ids += [None] * (max_depth + 1 - len(level_ids))
    
    # 生成每个层级的名称
    level_names = [node_map.get(id_) if id_ is not None else None for id_ in level_ids]
    
    # 把层级字段加到原行数据里
    for idx in range(max_depth + 1):
        row[f'Level{idx}_Id'] = level_ids[idx]
        row[f'Level{idx}_Name'] = level_names[idx]
    
    return row

# 应用函数到所有行，得到扁平化结果
flattened_df = df.apply(flatten_node, axis=1)

# 调整列顺序，让层级字段排在后面
base_columns = ['Id', 'name', 'parentId', 'path', 'depth']
level_columns = [col for col in flattened_df.columns if col not in base_columns]
flattened_df = flattened_df[base_columns + sorted(level_columns)]

# 打印结果（也可以导出成CSV/Excel）
print(flattened_df.to_markdown(index=False))

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