PySpark计算截止每日用户最近登录时间 统计两年内30天未访问账号次数

核心思路是通过用户相邻两次登录的时间区间直接关联全量日历，避免逐天遍历聚合，性能提升几个数量级。

步骤说明 #

• 首先预处理登录数据：将时间戳转为日期格式，对同一个用户同一天的多次登录做去重，只保留一条记录即可，减少后续计算量。
• 生成统计周期内的全量日历表：只需要生成过去两年共约730条日历记录即可，计算成本极低。
• 计算每个用户相邻两次登录的时间区间：按用户分组、登录日期排序，用窗口函数取每条登录记录对应的下一次登录日期，这样每条登录记录的有效覆盖区间就是登录日期后1天 ~ 下次登录日期前1天，这个区间内的所有日历日的最近登录时间都是当前这条记录的登录日期。
• 关联日历表得到最终结果：将带相邻登录区间的登录表和日历表做范围join，直接得到每个用户、每个日历日对应的最近一次登录时间。

代码示例（PySpark） #

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

# 1. 登录数据预处理：转日期、去重同用户同天登录
login_clean_df = login_df \
    .withColumn("login_date", F.to_date(F.col("data"))) \
    .dropDuplicates(["loginuser", "login_date"])

# 2. 生成过去两年的全量日历表
stat_start_date = F.date_sub(F.current_date(), 365 * 2)
stat_end_date = F.current_date()
calendar_df = spark.sql(f"""
    SELECT explode(sequence(to_date('{stat_start_date}'), to_date('{stat_end_date}'), interval 1 day)) AS calendar_day
""")

# 3. 为每个用户的登录记录补全下一次登录时间
user_login_window = Window.partitionBy("loginuser").orderBy("login_date")
login_with_next_df = login_clean_df \
    .withColumn("next_login_date", F.lead("login_date", 1, F.to_date(F.lit("9999-12-31"))).over(user_login_window))

# 4. 范围关联得到最终结果
result_df = login_with_next_df.join(
    calendar_df,
    on=F.expr("calendar_day BETWEEN date_add(login_date, 1) AND date_sub(next_login_date, 1)"),
    how="inner"
).select(
    "loginuser",
    "calendar_day",
    F.col("login_date").alias("lastlogin_timestamp")
)

边界补充 #

如果需要统计用户首次登录之前的日期的最近登录状态，可以给每个用户补充一条虚拟的初始登录记录，登录日期设为远早于统计周期的日期（比如1970-01-01），下次登录日期设为用户的第一次实际登录日期即可。

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