PySpark炸开JSON不存在的key时如何保留字段并统一置空

现有PySpark代码处理JSON输入，对数组字段x执行explode_outer操作后提取子字段生成新字段，当输入文件为空或不存在JSON键x时，会抛出cannot resolve 'x' given input columns: []错误，需要实现即使key不存在，最终输出表的所有目标字段仍保留且全部填充为NULL。

原代码如下：

new_df = df \
    .withColumn("x", explode_outer(col("x"))) \
        .select(
            col("x.p").alias("xp"),
            col("x.q").alias("xq"),
            col("x.r.l.g").alias("xrlg"),
            col("x.r.m.f").alias("xrmf"),
            col("x.r.n").alias("xrn"),
            col("x.r.o").alias("xro"),
            col("x.r.s").alias("xrs"),
    )

步骤1：预先定义字段完整Schema #

提前定义好x字段的完整嵌套结构，保证即使字段不存在，也能通过类型声明匹配后续取数逻辑：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, explode_outer, lit
from pyspark.sql.types import StructType, StructField, ArrayType, StringType

# 按需调整各字段实际类型，数值类可换成IntegerType、DoubleType等
x_r_inner_schema = StructType([
    StructField("l", StructType([StructField("g", StringType(), nullable=True)]), nullable=True),
    StructField("m", StructType([StructField("f", StringType(), nullable=True)]), nullable=True),
    StructField("n", StringType(), nullable=True),
    StructField("o", StringType(), nullable=True),
    StructField("s", StringType(), nullable=True)
])

x_inner_schema = StructType([
    StructField("p", StringType(), nullable=True),
    StructField("q", StringType(), nullable=True),
    StructField("r", x_r_inner_schema, nullable=True)
])

# x是数组类型，外层包ArrayType
x_schema = ArrayType(x_inner_schema, nullable=True)

# 最终输出的目标字段列表
target_cols = ["xp", "xq", "xrlg", "xrmf", "xrn", "xro", "xrs"]

步骤2：兼容x字段不存在的场景 #

先检查原DataFrame是否包含x字段，不存在则手动添加值为NULL、类型匹配的x字段：

if "x" not in df.columns:
    df = df.withColumn("x", lit(None).cast(x_schema))

步骤3：兼容输入为空的场景 #

如果原DataFrame没有任何行，直接构造符合目标字段结构的空结果，避免后续操作报错：

if df.rdd.isEmpty():
    # 直接生成全为NULL的目标结构DataFrame
    new_df = spark.createDataFrame([], schema=StructType([StructField(c, StringType(), nullable=True) for c in target_cols]))
else:
    # 执行原逻辑，explode_outer对NULL数组会返回NULL行
    new_df = df \
        .withColumn("x", explode_outer(col("x"))) \
        .select(
            col("x.p").alias("xp"),
            col("x.q").alias("xq"),
            col("x.r.l.g").alias("xrlg"),
            col("x.r.m.f").alias("xrmf"),
            col("x.r.n").alias("xrn"),
            col("x.r.o").alias("xro"),
            col("x.r.s").alias("xrs"),
        )

注意事项 #

• 定义Schema时需和真实业务数据的字段类型保持一致，避免出现类型转换异常
• explode_outer本身兼容输入为NULL的数组，执行后只会返回一行所有字段为NULL的结果，不会抛出异常

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