Databricks Notebook大数量集下MERGE INTO转UPDATE/INSERT实现方法

整体逻辑与原脚本完全对齐：以ResponseRgBasketId、RequestTimestamp为联合匹配键，匹配命中目标表的行做全字段更新，未命中的源表行做全字段插入，适配大数据量下单条MERGE语句易出现的执行超时、小文件堆积、锁等待失败等问题。

实现步骤与对应代码 #

前置准备：动态生成字段映射 #

原MERGE脚本中使用的UPDATE SET *、INSERT *是引擎层面的便捷语法，拆分为独立语句时需要显式解析字段列表，避免字段错位写入：

# 定义联合匹配键
join_keys = ["ResponseRgBasketId", "RequestTimestamp"]
# 获取目标表全量字段
target_schema = spark.table(target_table_name).schema
target_all_cols = [field.name for field in target_schema.fields]
# 过滤出非匹配键的字段列表（用于UPDATE赋值，避免重复更新关联键触发语法报错）
non_join_cols = [col for col in target_all_cols if col not in join_keys]

# 拼接SQL片段
update_assign_clause = ", ".join([f"target.{col} = source.{col}" for col in non_join_cols])
insert_all_col_clause = ", ".join(target_all_cols)

第一步：执行匹配行的全字段更新 #

将源表与目标表按联合匹配键关联，把命中的目标表行更新为源表的字段值：

update_exec_sql = f"""
UPDATE {target_table_name} target
SET {update_assign_clause}
FROM {staging_table_name} source
WHERE
  source.ResponseRgBasketId = target.ResponseRgBasketId
  AND source.RequestTimestamp = target.RequestTimestamp
"""
spark.sql(update_exec_sql)

第二步：执行未匹配行的全字段插入 #

筛选源表中不存在于目标表的行（即两个匹配键的组合值在目标表无对应记录），全量插入到目标表中：

insert_exec_sql = f"""
INSERT INTO {target_table_name} ({insert_all_col_clause})
SELECT {insert_all_col_clause}
FROM {staging_table_name} source
WHERE NOT EXISTS (
  SELECT 1
  FROM {target_table_name} target
  WHERE
    source.ResponseRgBasketId = target.ResponseRgBasketId
    AND source.RequestTimestamp = target.RequestTimestamp
)
"""
spark.sql(insert_exec_sql)

注意事项 #

• 上述拆分操作适配Databricks上的Delta表、Parquet外表，所有写操作都会保留完整的事务日志，不会产生数据一致性问题
• 执行顺序必须严格遵循“先UPDATE、后INSERT”的顺序，避免出现重复写入、更新覆盖插入数据的逻辑偏差
• 如果staging表本身存在重复的匹配键组合，需要先对staging表按匹配键去重，否则会出现同一目标行被重复更新的问题

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