在Databricks中对比两个Delta表并实现值替换的方法

针对你的需求，最直接且高效的方式是使用Delta Lake的MERGE INTO语句（原子性UPSERT操作），或者通过PySpark DataFrame API实现左连接+条件转换。以下是两种实现方式：

方式一：使用Spark SQL的MERGE INTO（推荐） #

这种方式是Delta Lake原生支持的原子操作，能确保数据更新的一致性，适合生产环境使用。

MERGE INTO table1
USING table2
ON table1.c = table2.c
WHEN MATCHED THEN
  UPDATE SET
    d = table2.d,
    e = table2.e,
    f = table2.f,
    g = table2.g,
    h = table2.h,
    i = table2.i,
    j = table2.j,
    b = CASE WHEN table2.e = 'closed' THEN 'OK' ELSE table1.b END

关键说明： #

• MERGE INTO table1：指定要更新的目标表为table1
• USING table2：指定提供更新数据的源表为table2
• ON table1.c = table2.c：定义两表的匹配条件（通过列c关联）
• WHEN MATCHED THEN UPDATE SET：仅对匹配到的行执行更新：
  • 同步table2的d、e、f、g、h、i、j列覆盖table1对应列的原有值
  • 对列b做条件处理：如果table2的e值为'closed'，则将table1的b设为'OK'，否则保留table1原有的b值

方式二：使用PySpark DataFrame API #

如果需要更灵活的代码逻辑控制，可以用DataFrame的左连接+条件转换实现：

from pyspark.sql.functions import col, when

# 读取Delta表
table1 = spark.read.format("delta").load("/path/to/table1")
table2 = spark.read.format("delta").load("/path/to/table2")

# 执行匹配与更新逻辑
updated_table1 = table1.alias("t1") \
    .join(table2.alias("t2"), on="c", how="left") \
    .select(
        col("t1.a"),
        # 处理列b的条件更新
        when(col("t2.e") == "closed", "OK").otherwise(col("t1.b")).alias("b"),
        col("t1.c"),
        # 同步table2的列，未匹配时保留table1原值
        col("t2.d").otherwise(col("t1.d")).alias("d"),
        col("t2.e").otherwise(col("t1.e")).alias("e"),
        col("t2.f").otherwise(col("t1.f")).alias("f"),
        col("t2.g").otherwise(col("t1.g")).alias("g"),
        col("t2.h").otherwise(col("t1.h")).alias("h"),
        col("t2.i").otherwise(col("t1.i")).alias("i"),
        col("t2.j").otherwise(col("t1.j")).alias("j")
    )

# 将结果写回Delta表（根据需求选择mode：overwrite覆盖原表，append追加）
updated_table1.write.format("delta").mode("overwrite").save("/path/to/updated_table1")

关键说明： #

• 左连接（how="left"）保证保留table1的所有行，仅对匹配到table2的行执行更新
• 通过when函数实现列b的条件替换逻辑
• 其他需要同步的列使用otherwise确保未匹配到table2的行保留table1原有值

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