PySpark 多条件删除DataFrame行：同SID空值关联的跨记录规则实现

实现逻辑拆解 #

完全匹配规则的执行步骤如下：

• 先从原始数据中定位所有满足StartDate、EndDate、CID三字段全为null的记录，按SID提取这类记录对应的Date Added值，作为对应SID的截断日期；如果同一SID有多条这类空记录，取最晚的Date Added作为截断值
• 将截断日期通过SID关联回原始全量数据集
• 执行两层过滤：
  1. 直接删除所有三字段全为null的空值记录本身
  2. 对有截断日期的SID，只保留Date Added值大于等于截断日期的业务行；对没有截断日期的SID（即从未出现过三字段全空记录的SID，如示例中的1004、1005），所有行全部保留
• 删除关联过程产生的临时字段，得到最终结果

可直接运行的PySpark实现代码 #

from pyspark.sql import functions as F

# 1. 提取每个SID对应的截断日期
cutoff_df = (
    raw_df
    # 筛选三字段全为空的特殊空记录
    .filter(
        F.col("StartDate").isNull()
        & F.col("EndDate").isNull()
        & F.col("CID").isNull()
    )
    .select("SID", F.col("Date Added").alias("sid_cutoff_date"))
    # 若同一SID有多条空记录，取最晚的Date Added作为截断值
    .groupBy("SID")
    .agg(F.max("sid_cutoff_date").alias("sid_cutoff_date"))
)

# 2. 关联截断日期到原始数据
joined_df = raw_df.join(cutoff_df, on="SID", how="left")

# 3. 应用过滤规则生成结果
result_df = (
    joined_df
    # 过滤1：移除所有三字段全空的记录本身
    .filter(
        ~(
            F.col("StartDate").isNull()
            & F.col("EndDate").isNull()
            & F.col("CID").isNull()
        )
    )
    # 过滤2：有截断日期的SID只保留大于等于截断日期的行，无截断日期的SID全量保留
    .filter(
        F.col("sid_cutoff_date").isNull()
        | (F.col("Date Added") >= F.col("sid_cutoff_date"))
    )
    # 清理临时关联字段
    .drop("sid_cutoff_date")
)

注意：如果你的Date Added字段是字符串类型，需要先通过F.to_date(F.col("Date Added"))转换为标准日期类型再做值比较，避免字符串字典序对比带来的逻辑错误。

效果验证 #

以你提供的示例数据跑通后的结果完全匹配预期输出：

• SID=1001对应的截断日期为2021-03-29，因此会删掉3条Date Added=2021-03-26的旧业务行、1条三字段全空的特殊记录，仅保留2条Date Added=2021-03-29的业务行
• SID=1004、1005没有对应的截断记录，全部原始行正常保留

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