You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何获取PySpark DataFrame的源引用列列表?

如何获取PySpark DataFrame引用的源列列表?

给定一个PySpark DataFrame,能否通过编程方式获取该DataFrame实际引用的源数据列列表?

以下示例可以更清晰地说明需求:

import pyspark.sql.functions as func
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()
source_df = spark.createDataFrame(
    [("pru", 23, "finance"), ("paul", 26, "HR"), ("noel", 20, "HR")],
    ["name", "age", "department"],
)
source_df.createOrReplaceTempView("people")
sqlDF = spark.sql("SELECT name, age, department FROM people")
df = sqlDF.groupBy("department").agg(func.max("age").alias("max_age"))
df.show()

执行后输出结果:

+----------+--------+                                                           
|department|max_age |
+----------+--------+
|   finance|      23|
|        HR|      26|
+----------+--------+

在这个例子中,最终的df实际引用的源列是[department, age],如何通过编程方式获取这个列表?

我尝试过通过提取执行计划字符串来获取相关信息,代码如下:

df._sc._jvm.PythonSQLUtils.explainString(df._jdf.queryExecution(), "formatted")

返回的执行计划内容如下:

== Physical Plan ==
AdaptiveSparkPlan (6)
+- HashAggregate (5)
   +- Exchange (4)
      +- HashAggregate (3)
         +- Project (2)
            +- Scan ExistingRDD (1)


(1) Scan ExistingRDD
Output [3]: [name#0, age#1L, department#2]
Arguments: [name#0, age#1L, department#2], MapPartitionsRDD[4] at applySchemaToPythonRDD at NativeMethodAccessorImpl.java:0, ExistingRDD, UnknownPartitioning(0)

(2) Project
Output [2]: [age#1L, department#2]
Input [3]: [name#0, age#1L, department#2]

(3) HashAggregate
Input [2]: [age#1L, department#2]
Keys [1]: [department#2]
Functions [1]: [partial_max(age#1L)]
Aggregate Attributes [1]: [max#22L]
Results [2]: [department#2, max#23L]

(4) Exchange
Input [2]: [department#2, max#23L]
Arguments: hashpartitioning(department#2, 200), ENSURE_REQUIREMENTS, [plan_id=60]

(5) HashAggregate
Input [2]: [department#2, max#23L]
Keys [1]: [department#2]
Functions [1]: [max(age#1L)]
Aggregate Attributes [1]: [max(age#1L)#12L]
Results [2]: [department#2, max(age#1L)#12L AS max_age#13L]

(6) AdaptiveSparkPlan
Output [2]: [department#2, max_age#13L]
Arguments: isFinalPlan=false

虽然执行计划里包含了所需信息,但直接解析字符串既麻烦又不可靠。

换个角度问:有没有办法将执行计划作为可遍历、可探索的对象来获取,而非字符串?

更新:感谢@matt-andruff的回复,我尝试了以下代码:

df._jdf.queryExecution().executedPlan().treeString().split("+-"[-2])

返回内容:

' Project [age#1L, department#2]\n            '

这种方法虽然能提取到信息,但不够优雅且极易出错。我真正需要的是一种安全可靠、受官方API支持的方法来获取该列表,目前怀疑是否存在这样的方法。

内容的提问来源于stack exchange,提问作者jamiet

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.11 03:01:11