PySpark 2.0窗口函数：设置30天回溯统计的有效起始日期

嘿，我完全懂你的困扰——你已经搭好了30天滑动窗口的统计逻辑，但不想让那些回溯期凑不齐30天的早期数据（也就是2018-02-01之前的日期）影响结果，对吧？别担心，这里有一套清晰的解决方案，分步骤帮你搞定：

核心思路 #

你的核心需求是：只保留那些统计日期对应的30天回溯期完全落在你的数据集时间范围内（2018-01-01至2018-04-01）的记录。也就是说，第一个满足条件的日期是2018-02-01（因为2018-02-01往前推30天刚好是2018-01-02，而你的数据集从2018-01-01开始，所有回溯数据都存在）。

注意：不能直接过滤原始数据（比如删掉2018-02-01之前的记录），否则2018-02-01的窗口统计会缺失前面的30天数据。正确的做法是先完成全量窗口计算，再过滤掉那些回溯不完整的日期记录。

具体代码实现 #

假设你的原始DataFrame包含id（用户ID）、action_date（记录日期，格式为yyyy-MM-dd）、a（动作类型）三列。

1. 预处理日期字段 #

首先确保日期字段是PySpark可识别的DateType：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, to_date, count, lit, datediff
from pyspark.sql import Window

# 初始化SparkSession（如果还没初始化）
spark = SparkSession.builder.appName("30DayWindowStats").getOrCreate()

# 假设你的原始数据存在df中，先转换日期格式
df = df.withColumn("action_date", to_date(col("action_date"), "yyyy-MM-dd"))

2. 定义30天滑动窗口 #

按id和a（动作类型）分区，按日期排序，范围设为过去30天（用时间戳秒数计算，86400秒=1天）：

# 将日期转换为时间戳秒数，用于rangeBetween的数值范围计算
window_spec = Window.partitionBy("id", "a") \
                   .orderBy(col("action_date").cast("timestamp").cast("long")) \
                   .rangeBetween(-86400 * 30, 0)

# 计算每个ID、每个动作的30天滚动发生次数
df_with_counts = df.withColumn("action_30day_count", count("*").over(window_spec))

3. 过滤回溯不完整的日期 #

只保留action_date >= 2018-02-01的记录，或者用更灵活的日期差判断：

# 方法1：直接指定起始日期
final_df = df_with_counts.filter(col("action_date") >= "2018-02-01")

# 方法2：更灵活的日期差判断（适合数据集起始日期变化的情况）
# 计算当前日期与数据集起始日期的天数差，只保留差>=30的记录
final_df = df_with_counts.filter(datediff(col("action_date"), lit("2018-01-01")) >= 30)

4. 验证结果 #

你可以查看最终结果的日期范围，确认只包含2018-02-01及之后的记录：

final_df.select("action_date").distinct().orderBy("action_date").show()

关键注意点 #

• 不要提前过滤原始数据：如果删掉2018-02-01之前的记录，2018-02-01的窗口统计会缺失2018-01-01至2018-01-31的数据，导致结果不准确。
• 窗口范围的正确性：用rangeBetween结合时间戳秒数是PySpark 2.0中实现滑动时间窗口的标准方式，确保时间范围精确到30天。
• 分区字段的选择：如果需要按每个ID的每个动作单独统计，一定要把a加入partitionBy，否则会统计该ID所有动作的总次数。

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