基于其他DataFrame列的去重：PySpark/SQL中保留规则冲突的解决方法

首先需要明确冲突的处理逻辑，这里我们提供两种常用的冲突解决方案，你可以根据业务需求选择：

方案一：净得分优先级规则 #

统计每个名字的「保留次数」（作为name_a出现的次数）减去「被删除次数」（作为name_b出现的次数），最终得分>0则保留；得分<=0但无任何操作记录的名字（比如示例中的D）也保留。

PySpark 实现 #

from pyspark.sql import functions as F

# 1. 统计保留/删除次数
keep_counts = df2.groupBy("name_a").agg(F.count("*").alias("keep_count"))
remove_counts = df2.groupBy("name_b").agg(F.count("*").alias("remove_count"))

# 2. 关联原表并计算净得分
score_df = df1.join(keep_counts, df1.name == keep_counts.name_a, "left") \
    .join(remove_counts, df1.name == remove_counts.name_b, "left") \
    .select(
        df1.name,
        F.coalesce("keep_count", F.lit(0)).alias("keep_count"),
        F.coalesce("remove_count", F.lit(0)).alias("remove_count")
    ) \
    .withColumn("net_score", F.col("keep_count") - F.col("remove_count"))

# 3. 过滤保留结果
final_df = score_df.filter(
    (F.col("net_score") > 0) | 
    ((F.col("keep_count") == 0) & (F.col("remove_count") == 0))
).select("name")

final_df.show()

运行结果：

+----+
|name|
+----+
|   D|
+----+

示例中A、B、C的净得分均为0，因此被过滤；D无操作记录，保留。

Spark SQL 实现 #

# 创建临时视图
df1.createOrReplaceTempView("names")
df2.createOrReplaceTempView("name_pairs")

SELECT n.name
FROM names n
LEFT JOIN (
    SELECT name_a, COUNT(*) as keep_count
    FROM name_pairs
    GROUP BY name_a
) k ON n.name = k.name_a
LEFT JOIN (
    SELECT name_b, COUNT(*) as remove_count
    FROM name_pairs
    GROUP BY name_b
) r ON n.name = r.name_b
WHERE 
    (COALESCE(k.keep_count, 0) - COALESCE(r.remove_count, 0) > 0)
    OR (COALESCE(k.keep_count, 0) = 0 AND COALESCE(r.remove_count, 0) = 0)

方案二：保留优先规则 #

只要名字存在至少一次「保留指令」（作为name_a出现），就优先保留；无任何操作记录的名字也保留。

PySpark 实现调整过滤条件 #

final_df = score_df.filter(
    (F.col("keep_count") > 0) | 
    ((F.col("keep_count") == 0) & (F.col("remove_count") == 0))
).select("name")

运行结果会保留A、B、C、D，因为三者都有至少一次保留指令。

自定义冲突规则扩展 #

如果有业务专属的冲突逻辑，可直接修改过滤条件：

• 只要被标记过删除就移除：过滤条件改为COALESCE(r.remove_count, 0) == 0，最终仅保留D
• 按字母顺序优先保留：新增优先级字段，净得分相同时保留字母靠前的名字

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