基于Azure Synapse Analytics提取员工层级上级的方案咨询

针对你在Synapse SQL池里无法用递归CTE提取员工所有层级上级的问题，以下是几个实用的替代方案：

1. 预生成员工层级快照（推荐） #

如果公司组织架构不会频繁变动，最省心的方式是用支持递归的工具提前计算好全量员工的完整上级链，存成快照表供后续查询使用：

• 用Synapse Spark池编写递归逻辑遍历全量员工CSV，生成包含每个员工、对应上级、层级的完整数据集，然后写入SQL池的表中。
• 后续处理增量数据时，直接关联这个快照表就能快速拿到所有上级，无需实时递归计算。

示例Spark Python代码：

from pyspark.sql import functions as F

# 读取数据湖中的全量员工CSV
emp_df = spark.read.csv(
    "abfss://<容器名>@<存储账户名>.dfs.core.windows.net/路径/full_employees.csv",
    header=True,
    inferSchema=True
)

# 初始化层级数据，层级0代表员工自身
hierarchy_df = emp_df.withColumn("level", F.lit(0))
max_hierarchy_level = 10  # 根据实际组织架构设定最大层级
current_level = 1

# 递归遍历上级
while current_level <= max_hierarchy_level:
    # 关联当前层级的上级信息
    temp_df = hierarchy_df.join(
        emp_df,
        hierarchy_df["manager_id"] == emp_df["employee_id"],
        "left"
    ).select(
        hierarchy_df["employee_id"],
        emp_df["employee_id"].alias("manager_id"),
        emp_df["employee_name"].alias("manager_name"),
        F.lit(current_level).alias("level")
    ).filter(F.col("manager_id").isNotNull())

    # 合并到总层级表
    hierarchy_df = hierarchy_df.union(temp_df)
    current_level += 1

# 将结果写入Synapse SQL池的表
hierarchy_df.write.mode("overwrite").synapsesql("SQL池名.dbo.employee_hierarchy")

之后处理增量数据时，直接查询关联即可：

SELECT
    inc.employee_id,
    h.manager_id,
    h.manager_name,
    h.level
FROM
    OPENROWSET(
        BULK 'abfss://<容器名>@<存储账户名>.dfs.core.windows.net/路径/incremental_employees.csv',
        FORMAT = 'CSV',
        HEADER_ROW = TRUE
    ) AS inc
JOIN
    dbo.employee_hierarchy h ON inc.employee_id = h.employee_id
ORDER BY
    inc.employee_id, h.level;

2. 手动多表连接（适合固定短层级） #

如果你的组织架构层级固定且较少（比如最多3-4层），可以直接用多次LEFT JOIN来逐级关联上级：

SELECT
    inc.employee_id,
    inc.employee_name,
    -- 一级经理
    m1.employee_id AS manager_l1_id,
    m1.employee_name AS manager_l1_name,
    -- 二级经理
    m2.employee_id AS manager_l2_id,
    m2.employee_name AS manager_l2_name,
    -- 三级经理
    m3.employee_id AS manager_l3_id,
    m3.employee_name AS manager_l3_name
FROM
    OPENROWSET(
        BULK 'abfss://<容器名>@<存储账户名>.dfs.core.windows.net/路径/incremental_employees.csv',
        FORMAT = 'CSV',
        HEADER_ROW = TRUE
    ) AS inc
LEFT JOIN
    OPENROWSET(
        BULK 'abfss://<容器名>@<存储账户名>.dfs.core.windows.net/路径/full_employees.csv',
        FORMAT = 'CSV',
        HEADER_ROW = TRUE
    ) AS m1 ON inc.manager_id = m1.employee_id
LEFT JOIN
    OPENROWSET(
        BULK 'abfss://<容器名>@<存储账户名>.dfs.core.windows.net/路径/full_employees.csv',
        FORMAT = 'CSV',
        HEADER_ROW = TRUE
    ) AS m2 ON m1.manager_id = m2.employee_id
LEFT JOIN
    OPENROWSET(
        BULK 'abfss://<容器名>@<存储账户名>.dfs.core.windows.net/路径/full_employees.csv',
        FORMAT = 'CSV',
        HEADER_ROW = TRUE
    ) AS m3 ON m2.manager_id = m3.employee_id;

这种方式简单直接，但如果层级变动，需要手动修改SQL语句。

3. 用ADF数据流实现递归 #

如果你的流程已经在使用Azure Data Factory，可以用数据流的递归自连接来生成层级数据：

1. 创建数据流，读取全量员工CSV作为源。
2. 添加自连接转换，设置连接条件为源.manager_id == 自连接源.employee_id，同时添加level字段，每次递归递增1。
3. 设置递归终止条件（比如level达到设定的最大值，或者manager_id为NULL）。
4. 读取增量员工CSV，和生成的层级数据集关联，即可获取每个增量员工的所有上级。
   这种方式无需编写大量代码，适合可视化配置ETL流程的场景。

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