如何实现Databricks Delta Table仅在记录变更时更新而非全量覆盖？

现有Delta表数据 #

需求描述 #

每日运行脚本时，仅对目标Delta表执行以下操作：

• 仅当除load_ts外的业务字段发生变更时，更新对应记录并刷新load_ts为当前时间
• 无变更的记录保留原load_ts，不执行任何更新
• 输入DataFrame中的新记录（目标表无匹配主键）直接插入，load_ts设为当前时间

之前使用普通Delta Merge会更新所有匹配记录，不符合需求，可通过以下方案实现。

核心是在Merge的更新分支中添加业务字段变更判断，仅当字段实际变化时才执行更新。

步骤1：确定主键与业务字段 #

以db_name + table_name作为联合主键，用于匹配目标表与输入数据的记录；业务字段包括location、table_format、table_type，这些字段的变更需要触发更新。

步骤2：为输入DataFrame添加当前时间戳 #

先给输入数据新增当前时间字段，用于后续插入或更新时的load_ts：

from pyspark.sql.functions import current_timestamp

input_df_with_ts = input_df.withColumn("new_load_ts", current_timestamp())

步骤3：执行带条件的Merge操作 #

使用Delta Lake的Merge API，在更新分支中加入字段比对逻辑：

from delta.tables import DeltaTable

# 加载目标Delta表
target_table = DeltaTable.forPath(spark, "dbfs:/path/to/your/target/delta/table")

# 执行Merge操作
(target_table.alias("target")
 .merge(
    input_df_with_ts.alias("source"),
    # 主键匹配条件
    "target.db_name = source.db_name AND target.table_name = source.table_name"
 )
 # 匹配且业务字段有变更时才更新
 .whenMatchedUpdate(
    condition="""
        target.location != source.location OR
        target.table_format != source.table_format OR
        target.table_type != source.table_type
    """,
    set={
        "location": "source.location",
        "table_format": "source.table_format",
        "table_type": "source.table_type",
        "load_ts": "source.new_load_ts"
    }
 )
 # 无匹配时插入新记录
 .whenNotMatchedInsert(
    values={
        "db_name": "source.db_name",
        "table_name": "source.table_name",
        "location": "source.location",
        "table_format": "source.table_format",
        "table_type": "source.table_type",
        "load_ts": "source.new_load_ts"
    }
 )
 .execute()
)

关键说明 #

• 匹配逻辑：通过联合主键精准匹配目标表与输入数据的记录
• 更新触发：仅当业务字段（location/table_format/table_type）任意一个与目标表记录不一致时，才执行更新并刷新load_ts
• 无变更处理：匹配但业务字段无变化的记录，不会触发任何操作，原load_ts保持不变
• 插入逻辑：新记录直接插入，load_ts设为当前时间

进阶优化：处理NULL值比对 #

如果业务字段可能存在NULL值，直接用!=会导致比对结果不准确，可改用Spark SQL的IS DISTINCT FROM语法，修改更新条件为：

target.location IS DISTINCT FROM source.location OR
target.table_format IS DISTINCT FROM source.table_format OR
target.table_type IS DISTINCT FROM source.table_type

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