PySpark中筛选有后续点击的搜索查询行的实现咨询

要实现仅保留同一用户5分钟内有后续点击的搜索行，可以通过两种高效的PySpark方式实现：窗口函数（适合大数据量）或自连接（逻辑更直观）。以下是具体实现步骤：

前提假设 #

合并后的combined_df包含核心字段：

• user_id：用户唯一标识（必须存在，否则无法关联同一用户的搜索/点击事件）
• timestamp：事件时间（需为TimestampType，若为字符串需先转换）
• query：搜索关键词（搜索行非空，点击行空）
• doc_id：点击文档ID（点击行非空，搜索行空）

方法一：窗口函数（推荐大数据场景） #

利用窗口函数按用户分区，在搜索行的5分钟时间窗口内判断是否存在后续点击事件：

from pyspark.sql import Window
from pyspark.sql.functions import col, unix_timestamp, max, when, lit

# 确保timestamp为时间戳类型（若原数据是字符串，先转换）
combined_df = combined_df.withColumn("timestamp", col("timestamp").cast("timestamp"))

# 定义窗口：按用户分区，按时间戳秒数排序，窗口范围为当前行之后5分钟（300秒）
window_spec = Window.partitionBy("user_id") \
                   .orderBy(unix_timestamp("timestamp")) \
                   .rangeBetween(0, 300)

# 标记是否有后续点击，并生成valid字段
result_df = combined_df.withColumn(
    # 窗口内存在点击事件则返回1，否则0
    "has_follow_click",
    max(when(col("doc_id").isNotNull(), 1)).over(window_spec)
).withColumn(
    # 仅搜索行且有后续点击时，valid标记为True
    "valid",
    when(col("query").isNotNull() & (col("has_follow_click") == 1), lit(True)).otherwise(lit(False))
).filter(col("valid") == True)  # 过滤出符合条件的搜索行

方法二：自连接（逻辑直观，适合中小数据量） #

通过将搜索行与点击行按用户和时间范围关联，直接筛选出有匹配点击的搜索行：

from pyspark.sql.functions import col, expr, lit, count

# 提取纯搜索行和纯点击行
search_df = combined_df.filter(col("query").isNotNull())
click_df = combined_df.filter(col("doc_id").isNotNull())

# 左连接：匹配同一用户、点击时间在搜索后5分钟内的记录
valid_searches = search_df.join(
    click_df,
    (search_df.user_id == click_df.user_id) &
    (click_df.timestamp >= search_df.timestamp) &
    (click_df.timestamp <= search_df.timestamp + expr("INTERVAL 5 MINUTES")),
    how="left"
).groupBy(search_df["*"]).agg(
    count(click_df.doc_id).alias("click_count")  # 统计匹配的点击数
).filter(col("click_count") > 0).withColumn("valid", lit(True))

关键注意事项 #

1. 时间戳类型：确保timestamp字段是TimestampType，否则时间范围计算会出错。若为字符串，可通过to_timestamp(col("timestamp"), "yyyy-MM-dd HH:mm:ss")转换。
2. 窗口范围：rangeBetween(0, 300)基于unix_timestamp的秒数差值，确保时间范围精确到5分钟。
3. 去重处理：若同一搜索行对应多个点击事件，两种方法都会自动去重（窗口函数取max，自连接groupBy去重）。

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