Databricks PySpark：如何指定spark.sql()返回DataFrame的非空字段约束？

直接在spark.sql(...)调用时指定返回DataFrame的Schema是做不到的——Spark SQL会根据查询语句的结果自动推导Schema，不支持直接传入自定义Schema覆盖。但你完全不需要用那种高开销的RDD转换方案，以下两种低开销的方法可以实现需求：

方法一：修改Schema元数据（无额外计算开销） #

Spark DataFrame的Schema本质是元数据，你可以直接构造新的StructType，将需要的列标记为nullable=false，再基于原DataFrame创建新的DataFrame（无需转换为RDD）：

from pyspark.sql.types import StructType, StructField

# 执行原SQL查询
original_df = spark.sql("SELECT id, name, age FROM your_table")
original_schema = original_df.schema

# 构造新Schema，指定目标列为非空
new_fields = []
for field in original_schema.fields:
    # 这里指定需要设置非空的列名，比如id和name
    if field.name in ("id", "name"):
        new_fields.append(StructField(field.name, field.dataType, nullable=False))
    else:
        new_fields.append(field)
new_schema = StructType(new_fields)

# 创建带有非空约束的新DataFrame（直接复用原DataFrame的执行计划，无额外开销）
df_with_non_null = spark.createDataFrame(original_df, schema=new_schema)

这种方式只是修改元数据，不会触发任何数据计算，性能开销可以忽略。

方法二：SQL查询+元数据修改（保证数据实际非空） #

如果需要确保返回的数据确实没有空值，可先在SQL语句中过滤空值，再修改Schema标记非空：

# 先通过SQL过滤空值
filtered_df = spark.sql("""
    SELECT id, name, age
    FROM your_table
    WHERE id IS NOT NULL AND name IS NOT NULL
""")

# 复用Schema修改逻辑
original_schema = filtered_df.schema
new_fields = [
    StructField(f.name, f.dataType, nullable=False) if f.name in ("id", "name") else f
    for f in original_schema.fields
]
new_schema = StructType(new_fields)

df_with_non_null = spark.createDataFrame(filtered_df, schema=new_schema)

这样既保证了数据层面无空值，又在元数据层面标记了非空约束，后续的Spark优化器可以利用这个约束生成更高效的执行计划。

注意：不要使用spark.createDataFrame(df.rdd, new_schema)的方案——转换为RDD会丢失Spark的优化信息，导致性能下降，直接传入原DataFrame才是低开销的正确方式。

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