PySpark动态展开嵌套列：AWS Glue环境替换低效for循环方案咨询

性能问题原因 #

你原来的for循环+逐次union写法性能差的核心原因是：每遍历一个属性就会全量扫描一次源数据，如果properties下有N个属性，源数据就会被重复读取N次，同时多次union也会让执行计划变得臃肿，在大数据量下性能衰减非常明显。

下面提供两种符合map/reduce风格的优化实现，性能比原写法提升2~N倍不等（取决于属性数量）： #

方案1：reduce合并子查询（兼容原逻辑，改动最小） #

直接用functools.reduce代替for循环逐次union，所有属性的子查询会并行生成后一次性合并，避免执行计划冗余：

from functools import reduce
from pyspark.sql import functions as F

# 1. 获取properties下的所有属性名
prop_cols = df_ex.select("properties.*").columns

# 2. 用map生成每个属性的展开后DataFrame
def process_prop(prop):
    return df_ex.select(
        "uniqueId",
        "modifiedDate",
        F.lit(prop).alias("propertyName"),
        F.explode(f"properties.{prop}.versions").alias("version")
    ).select(
        "uniqueId",
        "modifiedDate",
        "propertyName",
        F.col("version.name").alias("name"),
        F.col("version.value").alias("value"),
        F.col("version.timestamp").alias("timestamp")
    )

prop_dfs = map(process_prop, prop_cols)

# 3. 用reduce一次性合并所有DataFrame
final_df = reduce(lambda df1, df2: df1.unionByName(df2), prop_dfs)

# 验证结果
final_df.show(truncate=False)

方案2：单趟扫描实现（性能最优，推荐大数据量使用） #

直接将properties结构体转为键值对数组，只做一次explode即可完成所有属性的展开，源数据仅扫描1次，性能不受属性数量影响：

from pyspark.sql import functions as F

# 1. 把properties结构体转成 [struct(属性名, versions)] 格式的数组
prop_arr = F.array(*[
    F.struct(
        F.lit(prop).alias("propertyName"),
        F.col(f"properties.{prop}.versions").alias("versions")
    ) for prop in df_ex.select("properties.*").columns
])

# 2. 先展开属性数组，再展开每个属性下的versions数组
final_df = df_ex.withColumn("props", F.explode(prop_arr))\
    .withColumn("version", F.explode("props.versions"))\
    .select(
        "uniqueId",
        "modifiedDate",
        F.col("props.propertyName").alias("propertyName"),
        F.col("version.name").alias("name"),
        F.col("version.value").alias("value"),
        F.col("version.timestamp").alias("timestamp")
    )

# 验证结果
final_df.show(truncate=False)

额外优化建议（适配AWS Glue环境） #

• 如果你的源数据是分区存储的，可提前加pushdown_predicate过滤不需要的分区，减少扫描数据量
• 数据量特别大时，可以将explode替换为explode_outer避免空数组导致整行数据丢失
• 不需要额外缓存中间表，以上两种实现都是窄依赖，Spark会自动优化执行计划

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