如何在PySpark中获取Column对象对应的标准SQL语句

方案1：直接调用Scala侧原生sql方法（推荐） #

你在Scala中用到的my_col.expr.sql方法完全可以通过PySpark的Java网关直接调用，不需要额外依赖，代码如下：

from pyspark.sql.functions import col, expr

# 构造示例Column对象
my_col = col("example_column_{placeholder}").isin("A", "B")

# 调用Scala侧Expression的sql方法
standard_sql = my_col._jc.expr().sql()
print(standard_sql)
# 输出：(example_column_{placeholder} IN ('A', 'B'))

该方法的原理：

• PySpark的Column对象的_jc属性是Scala侧Column类的代理实例
• 调用.expr()会拿到Scala侧的Expression实例
• 调用.sql()即可获取符合ANSI SQL规范的表达式字符串，字符串字面量会自动带单引号

后续替换占位符的逻辑可以正常运行：

# 替换占位符
rendered_sql = standard_sql.format(placeholder="user_type")
# 执行表达式
df = df.withColumn("is_valid_type", expr(rendered_sql))

该方案兼容Spark 2.3及以上所有版本，是性能和准确性最优的方案。

方案2：低版本Spark兼容方案（仅适用于Spark<2.3） #

如果你的Spark版本低于2.3，没有内置sql方法，可以用临时查询的方式间接提取SQL：

def get_column_sql(col_obj):
    # 获取SparkSession实例
    from pyspark.sql import SparkSession
    spark = SparkSession.getActiveSession()
    # 构造空DataFrame做临时查询
    empty_df = spark.range(0).select(col_obj.alias("tmp"))
    # 从执行计划中提取标准SQL
    return empty_df._jdf.queryExecution().logical().expressions().head().sql()

该方案性能略低于方案1，但结果准确性和方案1一致。

内容的提问来源于stack exchange，提问作者Lukáš Langr