如何在PySpark中获取表/数据依赖关系？

1. 解析DataFrame的LogicalPlan #

Spark的DataFrame底层基于LogicalPlan执行，其中包含了明确的数据源节点，可以通过遍历计划结构提取依赖表名，这是最直接可靠的方式。

示例代码：

from pyspark.sql.catalyst.plans.logical import LogicalPlan

def extract_dependent_tables(df):
    dependencies = set()
    
    def traverse_plan(plan: LogicalPlan):
        # 匹配TableScan节点，提取库名和表名
        if hasattr(plan, 'tableName'):
            db = plan.database or df.sparkSession.catalog.currentDatabase()
            full_table = f"{db}.{plan.tableName}" if db else plan.tableName
            dependencies.add(full_table)
        # 递归遍历所有子节点
        for child in plan.children:
            traverse_plan(child)
    
    traverse_plan(df._jdf.queryExecution().logical())
    return dependencies

# 用法示例
target_df = spark.sql("SELECT * FROM sales.order JOIN sales.user ON order.user_id = user.id")
print(f"目标表依赖: {extract_dependent_tables(target_df)}")

这个方法适配SQL和DataFrame API构建的所有查询，能精准提取直接依赖的表。

2. 利用Spark Catalog系统元数据 #

如果表是持久化到Catalog中的（通过CREATE TABLE创建），可以查询系统元数据视图获取依赖，但仅适用于有明确约束或已注册的表：

示例SQL：

-- 查询表的直接关联依赖（Spark 3.0+支持）
SELECT 
    table_name,
    referenced_table_name
FROM 
    information_schema.table_constraints
WHERE 
    constraint_type = 'FOREIGN KEY'

若表是ETL生成的无约束表，可通过SHOW CREATE TABLE [table_name]解析建表语句中的数据源，但这种方式可靠性较低，需自行处理复杂语法。

3. 全局监听查询执行（批量场景适配） #

如果需要自动捕获所有作业的表依赖并写入日志，可自定义QueryExecutionListener：

示例代码：

from pyspark.sql.util import QueryExecutionListener

class DependencyLogger(QueryExecutionListener):
    def onSuccess(self, func_name, query_exec, duration):
        # 提取当前作业的依赖表
        deps = extract_dependent_tables(query_exec.sparkSession.createDataFrame(query_exec.executedPlan))
        # 写入日志文件
        with open("table_deps.log", "a", encoding="utf-8") as f:
            f.write(f"作业[{func_name}]依赖表: {','.join(deps)}\n")
    
    def onFailure(self, func_name, query_exec, exception):
        pass

# 注册监听器到SparkSession
spark = SparkSession.builder.appName("DepTracker").getOrCreate()
spark.sparkContext.addSparkListener(DependencyLogger())

该方案适合批量处理大量表的场景，无需手动逐个调用提取方法。

注意事项 #

• 上述方法默认提取直接依赖表，若需完整依赖链（比如依赖表本身还有上游依赖），需递归调用提取逻辑。
• 临时视图（createOrReplaceTempView）可通过LogicalPlan解析获取依赖，但无法通过系统元数据视图查询。
• 若使用自定义数据源，需调整traverse_plan中的节点判断逻辑，匹配对应数据源的LogicalPlan类型。

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