PySpark如何过滤StructType数组列内字段为null的元素

由于from_json方法会强制将所有字段的属性设置为可空，因此需要显式校验所有定义为非空的字段是否实际存在null值。普通列可直接通过case when语句判断列是否为null，针对数组类型列内嵌结构体的字段校验，可通过如下方式实现：

首先明确待校验的selections列结构定义：

StructField("selections", ArrayType(StructType([
    StructField("selectionId", IntegerType(), False),
    StructField("selectionIdXX", IntegerType(), False),
])))

本次校验目标为selections数组列内嵌结构体中的selectionId字段，只要数组中存在任意一个结构体的selectionId为null，就将对应行标记为InValid。

写法1：使用expr调用Spark SQL filter高阶函数 #

之前写法的核心问题是在expr的SQL字符串中错误嵌套了col()函数，SQL语法下直接引用列名、通过lambda参数访问嵌套字段即可。
filter高阶函数会返回数组中满足判断条件的所有元素，只要判断过滤后的数组长度是否大于0，就能确认数组中是否存在null值的违规数据，完整校验逻辑如下：

from pyspark.sql import functions as F

df = df.withColumn(
    "valid_status",
    F.when(
        # 普通非空列校验
        F.col("col1").isNull() | F.col("col2").isNull()
        # 数组嵌套结构体字段校验
        | (F.size(F.expr("filter(selections, item -> item.selectionId is null)")) > 0),
        "InValid"
    ).otherwise("Valid")
)

写法2：使用PySpark原生高阶函数API #

如果不想混用SQL表达式，也可以直接使用PySpark封装的filter高阶函数API，逻辑和上述写法完全等价：

from pyspark.sql import functions as F

df = df.withColumn(
    "valid_status",
    F.when(
        F.col("col1").isNull() | F.col("col2").isNull()
        | (F.size(F.filter(F.col("selections"), lambda item: item.getField("selectionId").isNull())) > 0),
        "InValid"
    ).otherwise("Valid")
)

注意事项 #

• expr内的代码遵循Spark SQL语法规则，不要在SQL字符串内调用DataFrame API的col()方法，否则会触发语法报错
• lambda表达式的参数名可自定义，访问结构体嵌套字段时，SQL写法用参数名.字段名格式，PySpark API写法用参数名.getField("字段名")格式
• 若需要同时校验数组内结构体的多个非空字段，直接在lambda的判断条件中用逻辑或拼接即可，例如校验selectionId和selectionIdXX两个字段时，判断条件可写为item -> item.selectionId is null or item.selectionIdXX is null
• 如果数组列本身允许为null，需要额外增加F.col("selections").isNull()的判断条件，避免高阶函数处理null值时返回不符合预期的结果

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