如何使用新Spark DataFrame对已有Spark DataFrame执行Upsert操作

你需要的操作属于典型的Upsert场景，核心逻辑为优先保留新DF的所有记录，同时补入旧DF中新DF未覆盖的主键对应记录，以下是两种常用实现方式：

方式一：原生DataFrame API实现（通用场景） #

无需依赖额外存储格式，适用于所有Spark版本：

• 第一步：为两个DF新增来源标识，标记新DF优先级高于旧DF

# 入参说明：old_df 为全量历史DF，new_df 为新增/更新DF，主键为 identity_number
from pyspark.sql import functions as F
from pyspark.sql.window import Window

# 新DF标记为1，优先级更高
new_with_flag = new_df.withColumn("_source_flag", F.lit(1))
# 旧DF标记为2，优先级更低
old_with_flag = old_df.withColumn("_source_flag", F.lit(2))

• 第二步：合并两个DF后按主键分组取优先级最高的记录

# 按字段名对齐合并两个DF
union_df = new_with_flag.unionByName(old_with_flag)

# 按主键分区，按优先级排序后取每个主键的第一条记录
window_spec = Window.partitionBy("identity_number").orderBy("_source_flag")
result_df = union_df.withColumn("rank", F.row_number().over(window_spec)) \
  .filter(F.col("rank") == 1) \
  .drop("rank", "_source_flag")

逻辑说明：新DF中存在的主键会覆盖旧DF的对应记录（实现更新），新DF独有的主键会直接保留（实现插入），旧DF中未被新DF覆盖的主键也会保留，完全符合需求。Scala语法逻辑完全一致，仅需调整语法格式即可。

方式二：Delta Lake Merge语法（大数据量推荐） #

如果你的历史数据已经存储为Delta Lake格式，直接用Merge语法性能提升非常明显，无需全量重算：

-- 先将新DF注册为临时视图
new_df.createOrReplaceTempView("new_increment_view")

-- 执行Merge操作
MERGE INTO delta.`/path/to/old_delta_table` t
USING new_increment_view s
ON t.identity_number = s.identity_number
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *

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