如何在Databricks Delta Lake上实现Spark DataFrame与Hive表的合并

针对你的Delta Lake表增量合并场景，最优方案是直接使用Delta Lake原生的MERGE INTO（upsert）能力，无需额外创建staging表、执行分区交换操作，天然适配你的有新增有更新的合并需求，数百万行量级的表性能完全可以满足。

实现代码 #

提供两种常用实现方式：

1. SQL写法（更简洁，适配schema完全一致的场景） #

// 将待合并的增量DataFrame注册为临时视图
df1.createOrReplaceTempView("incremental_data")

spark.sql("""
MERGE INTO db.table AS target
USING incremental_data AS source
-- 关联键为dept和id
ON target.dept = source.dept AND target.id = source.id
-- 匹配到的记录执行更新，全字段覆盖
WHEN MATCHED THEN UPDATE SET *
-- 未匹配到的记录执行新增，全字段插入
WHEN NOT MATCHED THEN INSERT *
""")

2. Scala Delta API写法 #

import io.delta.tables._

// 加载目标Delta表
val targetTable = DeltaTable.forName(spark, "db.table")

targetTable.as("target")
  .merge(
    df1.as("source"),
    "target.dept = source.dept AND target.id = source.id"
  )
  .whenMatched.updateAll()
  .whenNotMatched.insertAll()
  .execute()

性能优化建议 #

• 开启分区裁剪：如果你的目标表是分区表，且分区键包含关联键中的字段（比如dept是分区键），Delta会自动裁剪无关分区，仅扫描增量数据涉及的分区，大幅减少扫描数据量。如果分区键不在关联键中，也可以手动给合并条件加增量数据涉及的范围过滤，比如target.dept in ('Sales', 'Finance')。
• 关联键加Z-Order索引：给目标表的关联键dept、id创建Z-Order索引，可以让相同关联键的数据存储在相近位置，大幅减少匹配时扫描的文件数，命令如下：

  OPTIMIZE db.table ZORDER BY (dept, id)
  

• 广播小增量表：如果待合并的增量数据量远小于目标表，可以给增量表加广播hint，避免shuffle开销，SQL写法为USING /*+ BROADCAST(source) */ incremental_data AS source，Scala写法为df1.hint("broadcast")。
• 定期整理表文件：合并操作结束后可定期执行OPTIMIZE整理小文件，执行VACUUM清理过期快照，避免后续操作因为大量小文件导致性能下降。

对比原有分区交换方案的优势 #

• 流程更简单，无需维护staging表、申请分区交换权限，减少运维成本。
• IO开销更低，仅修改匹配到的目标数据文件，无需重写整个分区的全部数据，增量更新场景下性能提升显著。
• 天然ACID保证，合并过程原子性，不会出现中间状态的脏数据，合并过程中不影响原有表的正常查询。

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