使用PySpark计算用户最长连续访问日期段及天数

实现思路 #

核心是先通过日期偏移识别连续访问的分段，再统计每个分段的关键指标，最后按规则筛选每个用户的目标分段。

分步代码实现 #

假设原始数据集的DataFrame名为df，member_id为用户标识，AccessDate为访问日期。如果AccessDate不是日期类型，先转换：

from pyspark.sql import Window
import pyspark.sql.functions as F

# 转换日期类型（如果原始字段是字符串）
df = df.withColumn("AccessDate", F.to_date(F.col("AccessDate")))

1. 用户内排序并获取前一日日期
   按用户分组，对访问日期排序后，用lag函数获取上一次访问的日期：

   user_window = Window.partitionBy("member_id").orderBy("AccessDate")
   df_with_prev = df.withColumn("prev_access_date", F.lag("AccessDate", 1).over(user_window))
   

2. 标记连续访问分段
   通过判断当日与前一日的日期差是否为1，标记分段的起始点，再累加得到唯一的分段ID：

   df_segmented = df_with_prev.withColumn(
       "is_new_segment",
       F.when(F.datediff("AccessDate", "prev_access_date") == 1, 0).otherwise(1)
   ).withColumn(
       "segment_id",
       F.sum("is_new_segment").over(user_window.rowsBetween(Window.unboundedPreceding, 0))
   )
   

   相同segment_id的记录属于同一段连续访问。

3. 计算分段的连续天数与起止日期
   按用户和分段分组，统计每个分段的起始、结束日期和连续天数：

   segment_window = Window.partitionBy("member_id", "segment_id")
   df_segment_stats = df_segmented.withColumn(
       "start_date", F.min("AccessDate").over(segment_window)
   ).withColumn(
       "end_date", F.max("AccessDate").over(segment_window)
   ).withColumn(
       "continuous_days", F.datediff("end_date", "start_date") + 1
   ).select("member_id", "segment_id", "start_date", "end_date", "continuous_days").distinct()
   

4. 按规则筛选目标分段
   按连续天数降序、结束日期降序排序，取每个用户的第一条记录，就是符合要求的结果：

   rank_window = Window.partitionBy("member_id").orderBy(F.desc("continuous_days"), F.desc("end_date"))
   final_result = df_segment_stats.withColumn("rank", F.row_number().over(rank_window))\
       .filter(F.col("rank") == 1)\
       .drop("segment_id", "rank")
   

效果验证 #

• 无连续访问的用户：每个单独日期为一个分段，连续天数为1，最终取最晚的访问日期（起止日期相同）
• 多段连续的用户：自动筛选出连续天数最长的分段；若存在天数相同的分段，取结束日期最晚的，完全匹配你给出的示例场景。

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