Spark读取NDJSON时如何强制Schema的非空约束？

首先明确：Spark Schema中的nullable=False不会自动校验字段是否存在或非空，它只是元数据标记，仅用于告知Spark字段的逻辑约束，不会在读取阶段强制执行校验。要实现强制非空约束，需手动处理：

方法1：读取后过滤不符合约束的记录 #

读取文件后，直接过滤掉id为null的记录，或拆分出脏数据：

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType

spark = SparkSession.builder.appName("NDJSONNonNullCheck").getOrCreate()

# 定义预定义Schema，nullable=False仅作为元数据标记
schema = StructType([
    StructField("id", StringType(), False),
    StructField("name", StringType(), True)
])

# 读取NDJSON文件
df = spark.read.schema(schema).option("mode", "PERMISSIVE").json("path/to/your/file.ndjson")

# 方式1：直接过滤出符合约束的干净数据
clean_df = df.filter(df.id.isNotNull())

# 方式2：拆分干净数据与脏数据
from pyspark.sql.functions import col, when

df_with_flag = df.withColumn(
    "is_valid",
    when(col("id").isNotNull(), True).otherwise(False)
)
clean_df = df_with_flag.filter(col("is_valid"))
corrupt_df = df_with_flag.filter(~col("is_valid"))

方法2：自定义Row级校验逻辑 #

如果需要更复杂的校验（比如同时检查id非空且非空字符串），可以用UDF标记无效记录：

from pyspark.sql.functions import udf
from pyspark.sql.types import BooleanType

def is_record_valid(id_val):
    return id_val is not None and id_val.strip() != ""

validate_udf = udf(is_record_valid, BooleanType())

df_with_validity = df.withColumn("is_valid", validate_udf(col("id")))

方法3：Spark SQL表级约束（Spark 3.0+） #

Spark 3.0及以上版本支持表级约束，可创建临时表后添加非空校验：

df.createOrReplaceTempView("temp_table")

# 添加id非空约束
spark.sql("ALTER TABLE temp_table ADD CONSTRAINT id_not_null CHECK (id IS NOT NULL)")

# 开启约束检查配置
spark.sql("SET spark.sql.constraints.check.enabled=true")

# 查询时自动过滤不符合约束的记录
clean_df = spark.sql("SELECT * FROM temp_table")

关键说明 #

• Spark的mode参数（PERMISSIVE/FAILFAST/DROPMALFORMED）仅处理Schema类型不匹配的场景（比如字符串转数字失败），不会处理字段缺失导致的null值。
• nullable=False仅用于写入阶段判断是否允许null值，或作为Schema元数据参考，不参与读取阶段的校验逻辑。

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