Delta Lake中whenNotMatchedInsertAll引发无修改数据复制问题

问题原因 #

当merge语句包含whenNotMatchedInsertAll()分支时，Delta Lake的执行逻辑会默认对全量新数据集进行扫描，以检查是否存在未匹配旧数据的行需要插入。即使新旧数据完全一致、没有任何行需要更新或插入，Delta仍然会记录merge操作的元数据，日志中会显示全量数据扫描的痕迹（并非真的复制全量数据，而是操作流程的记录）。而注释掉whenNotMatchedInsertAll()后，merge仅处理匹配行的更新逻辑，当无符合更新条件的行时，不会触发全量扫描，因此没有日志生成。

解决方法 #

1. 提前过滤候选数据，仅在有变更时执行merge #

在执行merge前，先筛选出真正需要更新或插入的行，只有当候选集非空时才执行merge操作，避免无意义的全量扫描：

# 筛选需要更新的行：匹配旧数据且metadata_modified更新
update_candidates = newData.join(oldData, on="id", how="inner") \
    .filter("newData.metadata_modified > oldData.metadata_modified") \
    .select("newData.*")

# 筛选需要插入的行：未匹配旧数据的新行
insert_candidates = newData.join(oldData, on="id", how="left_anti")

# 仅当存在更新或插入需求时执行merge
if update_candidates.count() > 0 or insert_candidates.count() > 0:
    combined_candidates = update_candidates.union(insert_candidates)
    oldData.alias("oldData") \
        .merge(combined_candidates.alias("newData"), "oldData.id = newData.id") \
        .whenMatchedUpdateAll("newData.metadata_modified > oldData.metadata_modified") \
        .whenNotMatchedInsertAll() \
        .execute()

2. 启用Delta Lake优化配置（Databricks环境适用） #

开启spark.databricks.delta.merge.enableOptimizedMerge配置，该优化会减少merge操作的不必要扫描，避免无变更时生成冗余日志：

spark.conf.set("spark.databricks.delta.merge.enableOptimizedMerge", "true")

3. 优化数据分区与索引 #

为数据集按id或metadata_modified设置分区，或创建Z-Order索引，让merge操作仅扫描相关分区数据，降低全量扫描的开销：

# 按id分区（示例）
oldData.write.partitionBy("id").format("delta").save("/path/to/delta_table")
# 或创建Z-Order索引
from delta.tables import DeltaTable
delta_table = DeltaTable.forPath(spark, "/path/to/delta_table")
delta_table.optimize().zOrderBy("id")

内容的提问来源于stack exchange，提问作者Guilherme Torres Castro