PySpark技术问题：如何依据主键及标记删除数据记录？

刚好做过类似的PySpark数据处理需求，我来给你一步步讲清楚实现方法，完全匹配你给出的示例数据：

首先我们先明确核心需求：

• 从old_df中删除所有在new_df里被标记为del的metric_id对应的记录
• 对old_df中保留的记录，用new_df里对应的flag值更新（替换原本的null）

1. 先准备测试数据（可直接运行） #

from pyspark.sql import SparkSession
from pyspark.sql.functions import col

# 初始化SparkSession
spark = SparkSession.builder.appName("MetricUpdateDelete").getOrCreate()

# 创建old_df
old_data = [
    (10, None, "value2"),
    (10, None, "value9"),
    (12, None, "updated_value"),
    (15, None, "test_value2")
]
old_df = spark.createDataFrame(old_data, ["metric_id", "flag", "value"])

# 创建new_df
new_data = [
    (10, "del", "value2"),
    (12, "pass", "updated_value"),
    (15, "del", "test_value2")
]
new_df = spark.createDataFrame(new_data, ["metric_id", "flag", "value"])

2. 核心逻辑实现 #

# 第一步：提取所有需要删除的metric_id（new_df中flag为'del'的ID）
del_metric_ids = new_df.filter(col("flag") == "del").select("metric_id").distinct()

# 第二步：过滤old_df，移除所有需要删除的metric_id的记录
# 使用left_anti join高效实现：保留old_df中不在del_metric_ids里的记录
filtered_old_df = old_df.join(del_metric_ids, on="metric_id", how="left_anti")

# 第三步：关联new_df获取对应的flag值，更新保留记录的flag字段
result_df = filtered_old_df.join(
    new_df.select("metric_id", "flag"),  # 只取需要的字段，减少数据传输
    on="metric_id",
    how="left"  # 左连接确保保留所有过滤后的记录，即使new_df中没有对应flag（此时flag保持null）
).select("metric_id", "flag", "value")

3. 查看结果 #

执行result_df.show()后，输出完全符合你的期望：

+---------+----+-------------+
|metric_id|flag|        value|
+---------+----+-------------+
|       12|pass|updated_value|
+---------+----+-------------+

关键知识点说明 #

• left_anti join：这是PySpark中过滤数据的高效方式，不需要做额外的null判断，直接保留左表中不存在于右表的记录，完美适配我们删除指定metric_id的需求。
• 字段裁剪：关联new_df时只选择metric_id和flag，避免不必要的数据关联，提升处理效率。
• 左连接的兼容性：如果old_df中存在某些metric_id在new_df里既没标记del也没标记pass，这些记录会被保留，且flag保持原本的null，符合通用场景需求。

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