如何在Databricks(PySpark)中用MERGE INTO处理含空值的源数据写入Delta表

针对源数据null被推断为StringType导致与目标表复杂类型不兼容的问题，核心解决方案是从根源上让源数据Schema与目标表完全对齐，无需手动逐个转换列类型，具体步骤如下：

1. 复用目标表Schema读取源JSON #

Spark自动推断JSON Schema时，会将孤立的null默认识别为StringType，这是问题的核心。解决办法是读取源数据时强制使用目标表的Schema，让Spark直接将源中的null解析为对应列的类型（包括嵌套复杂类型）。

示例代码（Scala）：

// 获取目标表的Schema
val targetSchema = spark.table("your_target_delta_table").schema

// 用目标Schema读取源JSON
val sourceDF = spark.read
  .schema(targetSchema)
  .json("path/to/your/source/json/files")

这样处理后，源数据中ID=47的Nested列null会被正确识别为ArrayType(StructType)的null，而非StringType，从根源上避免类型不匹配。

2. 自动同步新增列（适配Schema演化需求） #

因为需要支持自动更新Schema，若源数据存在目标表没有的列，需先同步Schema再执行MERGE：

// 将源数据写入临时Delta表，开启Schema合并
sourceDF.write
  .mode("append")
  .option("mergeSchema", "true")
  .saveAsTable("temp_source_table")

// 自动将临时表的新增列同步到目标表
spark.sql("ALTER TABLE your_target_delta_table ADD COLUMNS (SELECT * FROM temp_source_table LIMIT 0)")

这一步利用Delta的Schema演化能力，自动识别并添加新增列，无需手动定义。

3. 动态生成MERGE INTO的SET语句 #

针对数百列的场景，无需手动编写SET子句，可通过遍历目标表列动态生成：

val primaryKey = "ID" // 替换为你的主键字段
val targetCols = targetSchema.fields.map(_.name).filter(_ != primaryKey)
val updateSetClause = targetCols.map(col => s"target.${col} = source.${col}").mkString(", ")

// 执行MERGE INTO
spark.sql(s"""
  MERGE INTO your_target_delta_table AS target
  USING sourceDF AS source
  ON target.$primaryKey = source.$primaryKey
  WHEN MATCHED THEN UPDATE SET $updateSetClause
  WHEN NOT MATCHED THEN INSERT *
""")

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

手动转换类型或动态生成SET语句时，无法改变Spark最初读取源数据时的类型推断结果——源数据的null已经被标记为StringType，后续转换无法彻底解决嵌套复杂类型的类型冲突。而复用目标Schema读取源数据是从数据加载阶段就强制对齐类型，从根源上解决问题。

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