如何获取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
相关产品推荐
相关产品推荐

