如何在PySpark结构化流作业中捕获被丢弃的重复事件

问题描述 #

我有一个PySpark流作业，通过session_id去除重复事件，设置了30分钟的水印窗口，代码片段如下：

unique_df = df.withColumn("timestamp", current_timestamp()).dropDuplicates(session_id).withWatermarking("timestamp", 30)

存在两个需求：

1. 捕获上述代码中所有被丢弃的事件，尝试过exceptAll和left_anti join但因流作业特性无法生效：

dropped_df = df.except(unique_df)

以及

dropped_events = df.join(unique_df, on=session_id, how="left_anti")

2. 若session_id=abc123在9:00首次出现，11:00因故障再次出现（超出30分钟水印窗口），需要标记并单独捕获该类重复事件。

解决方案 #

需求1：捕获所有被丢弃的重复事件 #

流处理中except/left_anti join失效的核心原因是水印的状态清理机制与流数据的增量处理特性。可通过分组聚合标记重复事件的方式实现：

1. 为每条数据添加处理时间戳，分组计算每个session_id的首次出现时间：

from pyspark.sql import functions as F

# 新增处理时间戳字段
df_with_ts = df.withColumn("processing_ts", F.current_timestamp())

# 按session_id分组，获取每个会话的首次处理时间
session_first_occur = df_with_ts.groupBy("session_id")\
    .agg(F.min("processing_ts").alias("first_processing_ts"))

# 关联原数据集，标记是否为重复事件
marked_df = df_with_ts.join(session_first_occur, on="session_id", how="inner")\
    .withColumn("is_duplicate", F.expr("processing_ts > first_processing_ts"))

2. 拆分唯一事件与被丢弃的重复事件：

# 提取唯一事件（首次出现的记录），并设置水印
unique_df = marked_df.filter(F.col("is_duplicate") == False)\
    .withWatermark("processing_ts", "30 minutes")\
    .drop("first_processing_ts", "is_duplicate")

# 捕获所有被丢弃的重复事件
dropped_df = marked_df.filter(F.col("is_duplicate") == True)\
    .drop("first_processing_ts")

这种方式通过聚合标记替代直接集合操作，适配流处理的增量特性，可准确捕获所有后续重复事件。

需求2：捕获超出水印窗口的过期重复事件 #

要区分窗口内重复与过期重复，需结合水印状态与时间差判断：

1. 在已标记重复的数据集基础上，添加过期判断逻辑：

# 基于唯一事件的水印，关联回标记数据集
final_marked_df = marked_df.join(
    unique_df,  # 已设置水印的唯一事件数据集
    on="session_id",
    how="left"
).withColumn(
    "is_expired_duplicate",
    F.expr("is_duplicate AND (processing_ts > first_processing_ts + interval 30 minutes)")
)

2. 拆分并捕获过期重复事件：

# 单独捕获超出30分钟窗口的重复事件
expired_dropped_df = final_marked_df.filter(F.col("is_expired_duplicate") == True)

# 窗口内的普通重复事件
normal_dropped_df = final_marked_df.filter(F.col("is_duplicate") & ~F.col("is_expired_duplicate"))

这里利用水印的状态保留机制，结合时间差计算判断重复事件是否超出有效窗口，实现精准分类捕获。

关键注意点 #

• 优先使用事件时间（业务自带的时间戳）而非处理时间，避免因系统延迟导致窗口判断误差。
• 流处理状态需合理配置，水印会自动清理过期状态，避免内存溢出。
• 若数据源存在乱序，需在设置水印时指定乱序容忍度。

内容的提问来源于stack exchange，提问作者boring-coder