PySpark中合并两个DataFrame：增量更新现有数据并插入新数据

看起来你已经理清了核心需求：以Item Id和item的组合作为关联键，合并两个DataFrame，重复条目累加count，新增条目直接保留。你的思路是正确的，但可以优化实现方式，让代码更简洁且性能更好。

先明确你的数据场景：

原始数据 #

DF1（主表）：

Item Id | item  | count
---------------------------
1       | item 1| 2
2       | item 2| 3
1       | item 3| 2
3       | item 4| 5

DF2：

Item Id | item  | count
---------------------------
1       | item 1| 2
3       | item 4| 2
4       | item 4| 4
5       | item 5| 2

期望结果：

Item Id | item  | count
---------------------------
1       | item 1| 4
2       | item 2| 3
1       | item 3| 2
3       | item 4| 7
4       | item 4| 4
5       | item 5| 2

你的现有方案分析 #

你的实现思路是对的：通过全外连接合并两个表，填充缺失值后求和。这个方法能得到正确结果，但存在可以优化的地方——join操作的开销比直接合并再分组要高，尤其是数据量较大时。

补全后的你的代码大概是这样（假设你给DF2的count重命名为newcount）：

from pyspark.sql import functions as F

temp = df2.withColumnRenamed("count", "newcount")
temp1 = df1.join(temp, ["Item Id", "item"], "full_outer").na.fill(0, subset=["count", "newcount"])
result_df = temp1.groupBy("Item Id", "item").agg(F.sum(F.col("count") + F.col("newcount")).alias("count"))
result_df.show()

更高效的实现方案 #

其实我们可以利用Spark的union操作先合并两个结构一致的DataFrame，再按关联键分组求和，这样逻辑更直接，性能也更好：

PySpark 代码示例

from pyspark.sql import functions as F

# 合并两个DataFrame（结构一致时直接union）
combined_df = df1.union(df2)

# 按关联键分组，对count求和
result_df = combined_df.groupBy("Item Id", "item").agg(F.sum("count").alias("count"))

# 查看结果（排序后更接近期望格式）
result_df.orderBy("Item Id", "item").show()

Scala 代码示例

import org.apache.spark.sql.functions.sum

val combinedDf = df1.union(df2)
val resultDf = combinedDf.groupBy("Item Id", "item").agg(sum("count").alias("count"))

resultDf.orderBy("Item Id", "item").show()

为什么这个方案更好？ #

1. 性能更优：union是轻量级的合并操作，不需要像join那样做复杂的关联计算，数据量越大，性能差距越明显。
2. 代码更简洁：省去了重命名列、填充缺失值的步骤，逻辑一目了然——合并所有数据，再按关联键汇总计数。
3. 结果一致：完全满足你的需求：重复条目累加count，新增条目直接保留，原有主表条目也不会丢失。

运行这段代码后，你就能得到你期望的结果啦。

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