如何不使用RDD API将Spark DataFrame可空列转为非可空列？

问题描述 #

我有一个带有可空列的PySpark DataFrame，希望将其转换为另一个DataFrame并使该列变为非可空：

from pyspark.sql.types import (
    StringType,
    StructType,
    StructField,
)

s = StructType(
    [
        StructField("ColumnA", StringType(), True)
    ]
)

df = spark.createDataFrame([("Blah",)], schema=s)

df.printSchema()  # 该列可空

df2 = df.select(df.ColumnA)

df2.printSchema()  # 仍为可空（希望改为非可空）

已尝试添加IS NOT NULL过滤、转换为StringType()、直接设置df2.schema['ColumnA'].nullable = False，均无效，且无法使用rdd API。

更大需求：将带有可空列的Databricks Delta表，转换为另一个具有等效但非可空列的Delta表。

有效解决方案 #

1. 显式定义非空Schema，重新生成DataFrame #

先确保数据中无null值（可选但推荐），再用自定义非空Schema创建新DataFrame：

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

# 过滤掉null值，保证数据符合非空约束
filtered_df = df.filter(df.ColumnA.isNotNull())

# 定义非可空的目标Schema
non_null_schema = StructType([
    StructField("ColumnA", StringType(), nullable=False)
])

# 基于过滤后的数据创建新DataFrame
df2 = spark.createDataFrame(filtered_df.toPandas(), schema=non_null_schema)

注：toPandas()适合中小数据量，大数据量建议结合下方Delta表写入方案。

2. 针对Delta表的直接写入方案 #

如果目标是生成非可空列的Delta表，可在写入时直接指定目标Schema：

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

# 读取源Delta表并过滤null值
source_df = spark.read.format("delta").load("/path/to/source_delta_table") \
    .filter("ColumnA IS NOT NULL")

# 定义目标表的非空Schema
target_schema = StructType([
    StructField("ColumnA", StringType(), nullable=False)
])

# 写入新Delta表，指定非空Schema
source_df.write.format("delta") \
    .mode("overwrite") \
    .schema(target_schema) \
    .save("/path/to/target_delta_table")

写入完成后，目标Delta表的ColumnA字段将是非可空状态。

为什么之前的方法无效？ #

• 添加IS NOT NULL过滤：Spark的Schema推断不会自动将可空列转为非空，即使数据中无null
• 转换为StringType()：仅转换数据类型，不会修改字段的可空属性
• 直接修改schema['ColumnA'].nullable：Spark的Schema是不可变对象，直接修改不会生效

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