Spark读取Delta表如何设置非空约束并校验空值

针对你的需求，这里提供几种比转RDD更优雅的实现方式，按推荐优先级排序：

1. 为Delta表添加表级非空约束（首推） #

直接在Delta表层面定义非空约束，这样所有读取、写入该表的操作都会自动验证规则，无需在读取代码里重复处理：

# 为Foo_ID和Bar列分别添加非空约束
spark.sql("ALTER TABLE delta.`/path/to/foo/table` ADD CONSTRAINT foo_id_not_null CHECK (Foo_ID IS NOT NULL)")
spark.sql("ALTER TABLE delta.`/path/to/foo/table` ADD CONSTRAINT bar_not_null CHECK (Bar IS NOT NULL)")

添加约束后，任何违反非空规则的写入操作会直接报错；读取时如果数据存在空值，也会触发约束检查并抛出明确错误，从根源上保障数据合规性。

2. 读取时指定自定义Schema + 显式空值检查 #

如果无法修改目标Delta表的结构，可以在读取阶段直接指定非空Schema，再显式检查空值并抛出错误，避免转RDD带来的性能损耗：

from pyspark.sql.types import StructType, StructField, LongType, TimestampType

# 定义带有非空约束的Schema
my_schema = StructType([
    StructField('Foo_ID', LongType(), False),
    StructField('Bar', TimestampType(), False)
])

# 读取Delta表时直接应用自定义Schema
df = (spark
      .read
      .format("delta")
      .schema(my_schema)
      .load("/path/to/foo/table")
      .select("Foo_ID", "Bar")
)

# 检查指定列是否存在空值，发现则抛出错误
for col_name in ["Foo_ID", "Bar"]:
    null_count = df.filter(df[col_name].isNull()).count()
    if null_count > 0:
        raise ValueError(f"列 {col_name} 检测到 {null_count} 条空值记录，违反非空约束")

3. 无RDD转换的Schema修正（需先做空值检查） #

如果仅需要修正DataFrame的Schema可空性标记，且已确认数据无空值，可以直接基于原DataFrame创建新实例，无需转RDD：

from pyspark.sql.types import StructType, StructField, LongType, TimestampType

# 读取原始数据
df = (spark
      .read
      .format("delta")
      .load("/path/to/foo/table")
      .select("Foo_ID", "Bar")
)

# 必须先做空值检查，避免后续Schema修正后空值静默存在
for col_name in ["Foo_ID", "Bar"]:
    if df.filter(df[col_name].isNull()).count() > 0:
        raise ValueError(f"列 {col_name} 存在空值，不符合非空要求")

# 定义非空Schema并创建新DataFrame
my_schema = StructType([
    StructField('Foo_ID', LongType(), False),
    StructField('Bar', TimestampType(), False)
])
df = spark.createDataFrame(df.rdd, schema=my_schema)

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