如何在PySpark中实现基于连续30天规则的客户流失（Churn）计算

假设你的数据集包含Id（客户ID）、date（日期）、using（使用状态，值为'true'/'false'字符串或布尔类型）三列，以下提供两种SQL实现方案，解决连续30天using=false时标记churn=true的需求。

方案一：标记单条记录是否处于连续≥30天的流失区间 #

如果需要给每条记录标记：当该记录所在的连续using=false区间长度≥30天时，churn=true，否则为false。

WITH ranked_data AS (
    SELECT
        Id,
        date,
        -- 将字符串状态转成布尔类型（如果你的using是布尔类型可跳过此步）
        (using = 'false') AS is_not_using,
        -- 用累计求和分割连续的未使用区间：每遇到一次使用状态（using=true），分组ID+1
        SUM(CASE WHEN NOT (using = 'false') THEN 1 ELSE 0 END) OVER (PARTITION BY Id ORDER BY date) AS group_id
    FROM your_table
),
interval_stats AS (
    SELECT
        Id,
        group_id,
        MIN(date) AS interval_start,
        MAX(date) AS interval_end,
        -- 计算区间天数（包含首尾当天）
        DATE_DIFF(MAX(date), MIN(date), DAY) + 1 AS interval_days
    FROM ranked_data
    WHERE is_not_using
    GROUP BY Id, group_id
),
qualified_intervals AS (
    SELECT Id, group_id
    FROM interval_stats
    WHERE interval_days >= 30
)
SELECT
    rd.Id,
    rd.date,
    rd.is_not_using AS using,
    CASE WHEN qi.group_id IS NOT NULL THEN TRUE ELSE FALSE END AS churn
FROM ranked_data rd
LEFT JOIN qualified_intervals qi 
    ON rd.Id = qi.Id AND rd.group_id = qi.group_id
ORDER BY rd.Id, rd.date;

方案二：标记客户是否存在过连续30天流失 #

如果只要客户存在任意一段连续30天using=false的记录，就将该客户所有记录的churn标记为true。

WITH ranked_data AS (
    SELECT
        Id,
        date,
        (using = 'false') AS is_not_using,
        SUM(CASE WHEN NOT (using = 'false') THEN 1 ELSE 0 END) OVER (PARTITION BY Id ORDER BY date) AS group_id
    FROM your_table
),
interval_stats AS (
    SELECT
        Id,
        group_id,
        DATE_DIFF(MAX(date), MIN(date), DAY) + 1 AS interval_days
    FROM ranked_data
    WHERE is_not_using
    GROUP BY Id, group_id
),
churn_customers AS (
    SELECT DISTINCT Id
    FROM interval_stats
    WHERE interval_days >= 30
)
SELECT
    t.Id,
    t.date,
    t.using,
    CASE WHEN cc.Id IS NOT NULL THEN TRUE ELSE FALSE END AS churn
FROM your_table t
LEFT JOIN churn_customers cc ON t.Id = cc.Id
ORDER BY t.Id, t.date;

注意事项 #

1. 日期缺失处理：如果数据集存在客户某几天没有记录的情况，上述方案会把缺失日期排除在连续天数计算外。若需要将缺失日期视为using=false，需先补全客户的所有日期记录，以BigQuery为例：

   WITH all_dates AS (
       SELECT
           Id,
           date
       FROM (SELECT DISTINCT Id FROM your_table),
       UNNEST(GENERATE_DATE_ARRAY(
           (SELECT MIN(date) FROM your_table),
           (SELECT MAX(date) FROM your_table),
           INTERVAL 1 DAY
       )) AS date
   ),
   full_data AS (
       SELECT
           ad.Id,
           ad.date,
           COALESCE((t.using = 'false'), FALSE) AS is_not_using
       FROM all_dates ad
       LEFT JOIN your_table t ON ad.Id = t.Id AND ad.date = t.date
   )
   -- 后续用full_data替代your_table执行方案一/二的逻辑
   

2. 函数适配：不同SQL方言的日期函数略有差异，比如PostgreSQL用AGE(MAX(date), MIN(date))或DATE_PART('day', MAX(date) - MIN(date)) + 1计算天数，generate_series生成日期序列，可根据你的数据库调整。

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