PySpark合并两个DataFrame：交集行优先取第一个DataFrame值

两个待合并的DataFrame A、B Schema一致，包含email_address、subject_line两个字符串字段，合并规则为：

• 保留两个DataFrame中所有出现过的email_address
• 同一email_address在A、B中同时存在时，最终结果的subject_line取A中对应值
• 同一email_address仅在单个DataFrame中存在时，取对应DataFrame中的subject_line值

你目前自行实现的leftanti join + union代码逻辑是完全正确的：

df_B_minus_A = df_B.join(df_A, ["email_address"], "leftanti")
result = df_A.union(df_B_minus_A)

该写法语义清晰，理解成本低，常规数据量场景下可以直接使用。如果要追求更优性能或者更好的扩展性，可以参考以下两种实现方式：

方案1：全外连接 + coalesce（性能最优，适合大数据量场景） #

该方案仅需一次join操作即可完成逻辑，相比leftanti+union减少一次shuffle过程，数据量较大时性能优势明显：

from pyspark.sql import functions as F

result = df_A.alias("a")\
    .join(df_B.alias("b"), on="email_address", how="full_outer")\
    .select(
        "email_address",
        F.coalesce(F.col("a.subject_line"), F.col("b.subject_line")).alias("subject_line")
    )

核心逻辑：

• 全外连接会保留A、B中所有的email_address记录，不会丢数
• coalesce函数按顺序取值，优先返回A表的subject_line，当A表无对应记录时（即邮箱仅存在于B中），自动取B表的对应值
• 最终结果天然无重复email_address，不需要额外去重

方案2：窗口函数排序取优先级最高记录（扩展性最优，适合多字段场景） #

如果后续除了subject_line外还有更多字段需要遵循“A表优先”的取值规则，用窗口函数的方式维护成本更低，不需要逐个字段写取值逻辑：

from pyspark.sql import functions as F
from pyspark.sql.window import Window

# 标记数据来源优先级，A表优先级设为1（最高），B表设为2
df_a_tag = df_A.withColumn("source_priority", F.lit(1))
df_b_tag = df_B.withColumn("source_priority", F.lit(2))

all_union = df_a_tag.unionByName(df_b_tag)
# 按邮箱分组，按优先级升序排序取第一条
email_window = Window.partitionBy("email_address").orderBy(F.col("source_priority").asc())
result = all_union.withColumn("rn", F.row_number().over(email_window))\
    .filter(F.col("rn") == 1)\
    .drop("source_priority", "rn")

核心逻辑：

• 给两个表的数据打上来源优先级标记，A表优先级高于B表
• 合并全量数据后，按email_address分组，取每个组内优先级最高的第一条记录即可
• 后续新增字段不需要修改取值逻辑，直接复用分组取第一条的规则即可

选型参考 #

• 团队新人多、代码可读性优先：选你自己写的leftanti + union方案，逻辑直白无理解门槛
• 数据量极大、性能优先：选全外连接+coalesce方案，shuffle次数最少，运行效率最高
• 表字段多、后续规则可能扩展：选窗口函数方案，维护成本最低

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