PySpark使用array函数处理字段缺失结构体时如何自动补NULL

错误根因 #

Spark的array函数要求所有输入元素的数据类型完全一致，你当前d字段下的两个子结构体schema不同（一个包含double类型的d字段，另一个没有），不符合array函数的参数要求，且该函数没有内置自动补全缺失字段为NULL的配置项。

解决方法 #

有两种常用的实现方案：

方案1：手动补全缺失字段 #

直接给缺少d字段的子结构体补全对应类型的NULL值，再组装数组执行explode操作：

from pyspark.sql import functions as f
from pyspark.sql.types import DoubleType

df2 = inputDF.withColumn("d",
    f.struct(
        f.col("d.1").alias("1"),
        # 给d.2补全double类型的d字段，值为null
        f.struct(
            f.col("d.2.a"),
            f.col("d.2.b"),
            f.col("d.2.c"),
            f.lit(None).cast(DoubleType()).alias("d")
        ).alias("2")
    )
).withColumn("d1", f.explode(f.array("d.*").getField("c")))

方案2：统一cast为相同schema（更推荐） #

如果结构体字段较多，可以提前定义通用的结构体schema，直接将不同的子结构体cast为该schema，Spark会自动将缺失字段赋值为NULL：

from pyspark.sql import functions as f
from pyspark.sql.types import StructType, StructField, StringType, DoubleType

# 定义统一的子结构体schema
common_struct_schema = StructType([
    StructField("a", StringType(), nullable=True),
    StructField("b", StringType(), nullable=True),
    StructField("c", StringType(), nullable=True),
    StructField("d", DoubleType(), nullable=True)
])

df2 = inputDF.withColumn("d",
    f.struct(
        f.col("d.1").cast(common_struct_schema).alias("1"),
        f.col("d.2").cast(common_struct_schema).alias("2")
    )
).withColumn("d1", f.explode(f.array("d.*").getField("c")))

第二种方案适配性更强，后续如果新增其他结构不同的子结构体，只需统一cast为通用schema即可，不需要修改补全逻辑。

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