PySpark源结构频繁变动场景下JSON数组展开方案

你当前手动枚举结构体字段的写法在Schema变动时需要同步修改选列代码，以下几种写法不需要手动维护结构体内部的字段列表，适配Schema频繁变动的场景：

方案1：用.*语法自动平铺结构体字段 #

这是最简单的通用写法，不需要手动列ID/hst/gst这些结构体内部字段，代码会自动把explode后得到的结构体下所有字段平铺到顶层：

from pyspark.sql.functions import explode, col

# 1. 展开tax数组得到结构体列
explode_temp = sourceDF.withColumn("tax_detail", explode(col("tax")))

# 2. 自动选列：保留原有所有非数组的顶层列 + 平铺结构体所有子字段
result_df = explode_temp.select(
    # 自动保留除了被展开的tax数组之外的所有顶层列（比如当前场景的Date列，后续顶层加新字段也会自动带上）
    *[col(col_name) for col_name in sourceDF.columns if col_name != "tax"],
    # 自动平铺结构体下的所有字段，不管结构体内部加/减字段都不需要改代码
    col("tax_detail.*")
)

如果需要把Date重命名为输出格式里的大写DATE，只需要在选顶层列的时候加个别名即可，不影响通用逻辑。

方案2：省略中间列的精简写法 #

不需要提前注册explode的临时列，直接链式调用完成展开，逻辑更紧凑：

from pyspark.sql.functions import explode, col

result_df = sourceDF.select(
    # 保留除tax数组外的所有顶层列
    *[col(col_name) for col_name in sourceDF.columns if col_name != "tax"],
    # 直接explode数组并命名为临时结构体列
    explode(col("tax")).alias("tax_detail")
).select(
    # 保留已选的所有列，同时平铺结构体的所有子字段
    "*",
    col("tax_detail.*")
).drop("tax_detail") # 删掉临时结构体列，得到最终扁平结果

扩展：自定义字段过滤规则 #

如果后续结构体里出现不需要的字段，不需要修改全量选列逻辑，只需要加简单的过滤条件即可，比如要排除结构体里的临时测试字段test_flag：

# 动态获取结构体下的所有字段名
struct_field_list = explode_temp.schema["tax_detail"].dataType.fieldNames()
# 过滤掉不需要的字段
selected_struct_cols = [
    col(f"tax_detail.{field_name}") 
    for field_name in struct_field_list 
    if field_name != "test_flag"
]
# 代入选列逻辑即可
result_df = explode_temp.select(
    *[col(col_name) for col_name in sourceDF.columns if col_name != "tax"],
    *selected_struct_cols
)

以上写法的适配性：

• 当tax数组内的结构体新增/删除字段时，不需要修改代码，输出结果会自动同步字段变化
• 当源DataFrame顶层新增除tax外的普通列时，代码会自动把新列带到结果中，不需要手动补充选列逻辑
• 如果后续要展开其他数组结构体列，只需要把代码里的tax替换成对应列名即可复用

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