如何用PySpark将DataFrame数据写入Delta表并适配列数差异？

Delta Lake 原生支持Schema 演化（新增列），同时我们可以通过动态调整 DataFrame 结构适配现有 Delta 表的需求，完全能实现你描述的两种场景。以下是具体的 PySpark 实现步骤：

步骤1：获取现有 Delta 表元数据 #

先读取目标 Delta 表的 Schema，确保新增列时数据类型匹配：

from pyspark.sql.functions import lit
from delta.tables import DeltaTable

# 替换为你的 Delta 表路径或注册的表名
delta_table_path = "/path/to/your/delta_table"
delta_table = DeltaTable.forPath(spark, delta_table_path)
delta_schema = delta_table.schema
delta_columns = set(delta_schema.fieldNames())

步骤2：根据输入 DataFrame 动态调整结构 #

针对两种不同的输入场景，分别处理 DataFrame：

# incoming_df 为你要写入的输入 DataFrame
incoming_df = ...
incoming_columns = set(incoming_df.columns)

updated_df = None
if incoming_columns == {"A", "B", "C", "D"}:
    # 场景1：仅含4列，新增E列并设为Null（匹配Delta表的E列数据类型）
    e_field = next(field for field in delta_schema.fields if field.name == "E")
    updated_df = incoming_df.withColumn("E", lit(None).cast(e_field.dataType))
elif incoming_columns == {"A", "B", "C", "D", "E", "F"}:
    # 场景2：含6列，保留原结构，后续通过Schema演化新增F列
    updated_df = incoming_df
else:
    # 可根据需求扩展其他场景，或抛出异常终止流程
    raise ValueError("输入DataFrame的列结构不符合预期场景")

步骤3：写入 Delta 表并适配 Schema #

根据是否需要新增列，启用对应的写入选项：

write_options = {}
# 检查是否需要新增F列，启用Schema演化
if "F" in updated_df.columns and "F" not in delta_columns:
    write_options["mergeSchema"] = "true"

# 写入Delta表（默认用append模式添加新数据，可按需改为overwrite等）
updated_df.write.format("delta") \
    .mode("append") \
    .options(**write_options) \
    .save(delta_table_path)

关键说明 #

1. 场景1处理：通过lit(None).cast()确保新增的E列数据类型与Delta表完全一致，避免类型不兼容问题。
2. 场景2处理：Delta Lake 的mergeSchema=true选项会自动将新增的F列添加到Delta表的Schema中，无需手动修改表结构。
3. 写入模式：示例使用append模式添加新数据，若需要覆盖全表可改为mode("overwrite")，但需谨慎操作。

内容的提问来源于stack exchange，提问作者Harshith K R