PySpark动态生成SQL透视语句遇阻，求技术解决方案

我需要在PySpark中动态生成SQL表达式实现数据透视，因为indicator列的取值不确定，没法写死SQL。手动编写静态SQL和使用DataFrame的pivot方法都能成功实现透视，但动态生成的SQL执行失败。试了三种获取distinct indicator值的方式，生成的表达式结构看似正确，但就是无法正常运行，求解决。

问题代码 #

from pyspark.sql import functions as F

# 获取distinct indicator值的三种尝试
indicators = [i.indicator for i in covid_df.select('indicator').distinct().collect()]
indicators = covid_df.toPandas()["indicator"].unique()
indicators = covid_df.select(F.collect_set("indicator").alias("column")).first()["column"]

# 动态生成透视表达式
pivot_exprs = [F.sum(F.when(F.col("indicator") == ind, F.col("value"))).alias(ind) for ind in indicators]

# 构造动态SQL查询
query = """
    SELECT entity, iso_code, date, {}
    FROM covid
    GROUP BY entity, iso_code, date
    ORDER BY entity, date
""".format(", ".join(pivot_exprs))

# 执行查询
result = spark.sql(query)
result.show()

数据集示例 #

entity,iso_code,date,indicator,value
Portugal,PRT,2020-07-28,Daily ICU occupancy,41.0
Portugal,PRT,2020-07-28,Daily ICU occupancy per million,3.992
Portugal,PRT,2020-07-28,Daily hospital occupancy,402.0
Portugal,PRT,2020-07-28,Daily hospital occupancy per million,39.14
Portugal,PRT,2020-07-29,Daily ICU occupancy,43.0
Portugal,PRT,2020-07-29,Daily ICU occupancy per million,4.187
Portugal,PRT,2020-07-29,Daily hospital occupancy,403.0
Portugal,PRT,2020-07-29,Daily hospital occupancy per million,39.237
Portugal,PRT,2020-07-30,Daily ICU occupancy,42.0
Portugal,PRT,2020-07-30,Daily ICU occupancy per million,4.089
Portugal,PRT,2020-07-30,Daily hospital occupancy,403.0
Portugal,PRT,2020-07-30,Daily hospital occupancy per million,39.237
Portugal,PRT,2020-07-31,Daily ICU occupancy,41.0

可行的静态实现示例 #

静态SQL实现 #

query = """
    SELECT entity, iso_code, date,
        SUM(CASE WHEN indicator = 'Daily ICU occupancy' THEN value END) AS `Daily ICU occupancy`,
        SUM(CASE WHEN indicator = 'Daily ICU occupancy per million' THEN value END) AS `Daily ICU occupancy per million`,
        SUM(CASE WHEN indicator = 'Daily hospital occupancy' THEN value END) AS `Daily hospital occupancy`,
        SUM(CASE WHEN indicator = 'Daily hospital occupancy per million' THEN value END) AS `Daily hospital occupancy per million`,    
        SUM(CASE WHEN indicator = 'Weekly new ICU admissions' THEN value END) AS `Weekly new ICU admissions`,
        SUM(CASE WHEN indicator = 'Weekly new ICU admissions per million' THEN value END) AS `Weekly new ICU admissions per million`,
        SUM(CASE WHEN indicator = 'Weekly new hospital admissions' THEN value END) AS `Weekly new hospital admissions`,
        SUM(CASE WHEN indicator = 'Weekly new hospital admissions per million' THEN value END) AS `Weekly new hospital admissions per million`   
    
    FROM covid
    GROUP BY entity, iso_code, date
    ORDER BY entity, date
"""

sqltest = spark.sql(query)
sqltest.show()

DataFrame pivot实现 #

from pyspark.sql.functions import col
    
pivotdf = covid_df.groupBy(["entity", "iso_code", "date"]).pivot("indicator").sum("value").sort(["entity", "date"])
pivotdf.limit(10).show()

尝试的三种获取indicator方式及生成的表达式 #

# 三种获取distinct indicator的方式
v1 indicators = [i.indicator for i in covid_df.select('indicator').distinct().collect()]
v2 indicators = covid_df.toPandas()["indicator"].unique()
v3 indicators = covid_df.select(F.collect_set("indicator").alias("column")).first()["column"]

# 生成的表达式
pivot_exprs = [F.sum(F.when(F.col("indicator") == ind, F.col("value"))).alias(ind) for ind in indicators]

生成的表达式结果：

[Column<'sum(CASE WHEN (indicator = Weekly new hospital admissions per million) THEN value END) AS `Weekly new hospital admissions per million`'>,
 Column<'sum(CASE WHEN (indicator = Weekly new ICU admissions) THEN value END) AS `Weekly new ICU admissions`'>,
 Column<'sum(CASE WHEN (indicator = Daily hospital occupancy per million) THEN value END) AS `Daily hospital occupancy per million`'>,
 Column<'sum(CASE WHEN (indicator = Daily hospital occupancy) THEN value END) AS `Daily hospital occupancy`'>,
 Column<'sum(CASE WHEN (indicator = Weekly new ICU admissions per million) THEN value END) AS `Weekly new ICU admissions per million`'>,
 Column<'sum(CASE WHEN (indicator = Daily ICU occupancy per million) THEN value END) AS `Daily ICU occupancy per million`'>,
 Column<'sum(CASE WHEN (indicator = Daily ICU occupancy) THEN value END) AS `Daily ICU occupancy`'>,
 Column<'sum(CASE WHEN (indicator = Weekly new hospital admissions) THEN value END) AS `Weekly new hospital admissions`'>]

解决方案 #

核心问题：动态生成的CASE语句中，字符串值没有被单引号包裹，Spark会把这些值识别为列名而非字符串常量，导致语法错误。

修正后的代码：

from pyspark.sql import functions as F

# 推荐用collect_set获取distinct值（大数据场景更高效，减少Driver端数据量）
indicators = covid_df.select(F.collect_set("indicator").alias("column")).first()["column"]

# 直接生成带单引号的SQL字符串片段，而非Column对象
pivot_exprs = [
    f"SUM(CASE WHEN indicator = '{ind}' THEN value END) AS `{ind}`"
    for ind in indicators
]

# 构造并执行SQL
query = f"""
    SELECT entity, iso_code, date, {', '.join(pivot_exprs)}
    FROM covid
    GROUP BY entity, iso_code, date
    ORDER BY entity, date
"""

result = spark.sql(query)
result.show()

补充说明 #

1. 不要直接拼接Column对象：Column的字符串输出不会给常量值加单引号，这是导致SQL语法错误的关键。
2. 优先使用collect_set：相比collect()或转Pandas，collect_set在大数据集下更高效，避免Driver端加载过多数据。
3. 别名用反引号包裹：确保indicator名称含空格或特殊字符时，SQL语法仍合法。

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