PySpark实现多ID列匹配行求和并保留无匹配行

针对你的需求，这里提供两种可行的实现方案，解决动态ID列的问题，同时兼顾效率：

方案一：Union + GroupBy 聚合 #

这个方案逻辑直观，实现简单，且能很好适配动态ID列的场景：

1. 先将两个DataFrame按列名合并（确保结构一致）；
2. 按动态ID列分组，对需要求和的列执行聚合。

代码示例 #

from pyspark.sql import functions as F

# 动态定义ID列和求和列
id_cols = ["C1", "C2"]
sum_cols = ["W1", "W2"]

# 合并两个DataFrame（unionByName确保列名对应，避免列顺序问题）
union_df = df_a.unionByName(df_b)

# 分组聚合求和
result_df = union_df.groupBy(id_cols).agg(
    *[F.sum(col).alias(col) for col in sum_cols]
)

效率说明 #

Union操作本身是轻量的宽依赖操作，不会触发shuffle；GroupBy的shuffle开销取决于数据总量和ID的基数。如果数据量极大，可以提前对两个DataFrame按ID列分区，减少后续shuffle的数据传输量，进一步优化性能。

方案二：Outer Join + 动态列求和 #

如果需要保留连接过程的中间状态，或者更灵活处理空值，可以选择全外连接后动态求和：

1. 按动态ID列执行全外连接；
2. 对每个求和列，将两个表的对应列（空值填充为0）相加；
3. 保留最终需要的列。

代码示例 #

from pyspark.sql import functions as F

id_cols = ["C1", "C2"]
sum_cols = ["W1", "W2"]

# 先给两个DataFrame的求和列重命名，避免连接后列名冲突
df_a_renamed = df_a.select(*id_cols, *[F.col(col).alias(f"{col}_a") for col in sum_cols])
df_b_renamed = df_b.select(*id_cols, *[F.col(col).alias(f"{col}_b") for col in sum_cols])

# 执行全外连接
joined_df = df_a_renamed.join(df_b_renamed, on=id_cols, how="outer")

# 动态处理每个求和列：空值转0后相加
for col in sum_cols:
    joined_df = joined_df.withColumn(
        col,
        F.coalesce(F.col(f"{col}_a"), F.lit(0)) + F.coalesce(F.col(f"{col}_b"), F.lit(0))
    )

# 保留ID列和求和后的最终列
result_df = joined_df.select(id_cols + sum_cols)

适用场景 #

适合需要对连接后的中间数据做额外处理的场景，比如查看某一侧表的缺失情况，但代码复杂度略高于方案一。

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