如何在Databricks中向Schema动态变化的Delta表插入数据

• insertInto 方法的匹配规则是按列位置顺序匹配，不参考列名，本身不支持Schema合并逻辑，无论你配置单操作的mergeSchema选项还是全局的spark.databricks.delta.schema.autoMerge参数，都不会生效。
• 你期望的按列名匹配、自动补NULL、自动扩展Schema的逻辑，需要使用按列名匹配的写入方法实现。

方法1：使用saveAsTable替代insertInto（最简便） #

saveAsTable按列名匹配写入，天然支持mergeSchema参数，完全满足你的需求，代码修改如下：

Python示例 #

loadfinaldf.write.format("delta")\
   .option("mergeSchema", "true")\
   .mode("append")\
   .saveAsTable("table")

Scala示例 #

loadfinaldf.write.format("delta")
  .option("mergeSchema", "true")
  .mode("append")
  .saveAsTable("table")

开启全局schema自动合并后可以省略单操作的mergeSchema参数：

spark.conf.set("spark.databricks.delta.schema.autoMerge", "true")
// 后续写入无需额外加mergeSchema选项
loadfinaldf.write.format("delta")
  .mode("append")
  .saveAsTable("table")

方法2：手动对齐Schema（兼容严格控制写入逻辑的场景） #

如果需要更可控的写入逻辑，可以先读取目标Delta表的现有Schema，给当前DataFrame补全缺失列、保留新增列后再写入，示例：

import org.apache.spark.sql.functions._
import io.delta.tables._

// 读取目标Delta表的Schema
val targetTable = DeltaTable.forName(spark, "table")
val targetCols = targetTable.toDF.columns.toSet
val currentCols = loadfinaldf.columns.toSet

// 补全当前DataFrame缺失的列，值填充为NULL
val finalDf = (targetCols -- currentCols).foldLeft(loadfinaldf)((df, col) => {
  df.withColumn(col, lit(null).cast(targetTable.toDF.schema(col).dataType))
})

// 写入时自动合并新增列
finalDf.write.format("delta")
  .option("mergeSchema", "true")
  .mode("append")
  .insertInto("table")

写入完成后查询Delta表，会自动新增E、F列，缺失的C列会自动填充为NULL，完全符合预期。

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