如何用Databricks SQL/PySpark计算用户每日连续登录天数

需求 #

使用Databricks SQL计算每个用户（first_install_id）在对应calendar_date的连续登录天数，若SQL无法实现可改用PySpark。

输入数据示例 #

user_id  calendar_date         
    John    2022-01-01  
    John    2022-01-10
    Mary    2022-01-01
    Steve   2022-01-03
    Mary    2022-01-03
    John    2022-01-02
    John    2022-01-03
    Mary    2022-01-04
    John    2022-01-04

期望输出示例 #

user_id  calendar_date consecutive_logins         
    John    2022-01-01          1
    John    2022-01-02          2
    John    2022-01-03          3
    John    2022-01-10          1
    Mary    2022-01-01          1
    Mary    2022-01-03          1
    Mary    2022-01-04          2
    Steve   2022-01-03          1

现有未完成代码 #

WITH dau as (
select
  distinct
  concat_ws('-', year, month, day) as calendar_date,
  first_install_id as fiu
  
    FROM
        production.events
    WHERE 1=1
    
      and concat_ws('-', year, month, day) >= '2023-11-01'
     
),

sorted_logins AS (
    SELECT
        calendar_date,
        fiu,
        DATEDIFF(calendar_date, LAG(calendar_date) OVER (PARTITION BY fiu ORDER BY calendar_date)) AS days_since_last_login
from dau
),

consecutive_logins AS (
    SELECT
        calendar_date,
        fiu,
        SUM(CASE WHEN days_since_last_login = 1 THEN 0 ELSE 1 END) OVER (PARTITION BY fiu ORDER BY calendar_date) AS consecutive_group
    FROM
        sorted_logins
) select * FROM consecutive_logins

完整解决方案代码 #

你已经完成了连续登录分组的核心逻辑，只需要在分组内生成递增序号即可得到连续登录天数。同时需要修正DATEDIFF的参数顺序（Databricks SQL中DATEDIFF语法为DATEDIFF(unit, start_date, end_date)），确保间隔天数计算正确：

WITH dau as (
select
  distinct
  concat_ws('-', year, month, day) as calendar_date,
  first_install_id as fiu
  
    FROM
        production.events
    WHERE 1=1
      and concat_ws('-', year, month, day) >= '2023-11-01'
     
),

sorted_logins AS (
    SELECT
        calendar_date,
        fiu,
        -- 修正DATEDIFF参数顺序，计算与上一次登录的间隔天数
        DATEDIFF(day, LAG(calendar_date) OVER (PARTITION BY fiu ORDER BY calendar_date), calendar_date) AS days_since_last_login
from dau
),

consecutive_groups AS (
    SELECT
        calendar_date,
        fiu,
        -- 生成连续登录分组：间隔不为1时，分组标识递增
        SUM(CASE WHEN days_since_last_login = 1 THEN 0 ELSE 1 END) OVER (PARTITION BY fiu ORDER BY calendar_date) AS consecutive_group
    FROM
        sorted_logins
),

final_result AS (
    SELECT
        fiu as user_id,
        calendar_date,
        -- 在每个用户的连续分组内，按日期排序生成连续登录天数
        ROW_NUMBER() OVER (PARTITION BY fiu, consecutive_group ORDER BY calendar_date) AS consecutive_logins
    FROM
        consecutive_groups
)
-- 按用户和日期排序输出，与示例格式一致
SELECT user_id, calendar_date, consecutive_logins
FROM final_result
ORDER BY user_id, calendar_date;

逻辑说明 #

1. dau CTE：去重得到每个用户每天的唯一登录记录，避免同一用户同一天多次登录影响计算。
2. sorted_logins CTE：按用户分组、日期排序，计算当前登录与上一次登录的间隔天数。
3. consecutive_groups CTE：通过累加标识生成连续登录的分组，每次间隔超过1天则开启新分组。
4. final_result CTE：在每个用户的每个连续分组内，按日期生成行号，行号即为该日期的连续登录天数。

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