播面

WITH daily_active AS (
    -- Step 1: 先按天进行买家去重，减少关联的数据量
    SELECT DISTINCT 
        purchase_date, 
        buyer_id
    FROM buyer_purchases
)
SELECT 
    t1.purchase_date,
    COUNT(DISTINCT t2.buyer_id) AS rolling_7d_active_buyers
FROM (
    -- 获取所有有交易的日期作为主轴
    SELECT DISTINCT purchase_date FROM buyer_purchases
) t1
JOIN daily_active t2 
  ON t2.purchase_date BETWEEN DATE_SUB(t1.purchase_date, 6) AND t1.purchase_date
GROUP BY 
    t1.purchase_date
ORDER BY 
    t1.purchase_date;

WITH daily_buyer_sets AS (
    -- Step 1: 聚合每天的买家并去重，存入 Array 释放行数限制
    SELECT 
        CAST(purchase_date AS DATE) AS purchase_date,
        collect_set(buyer_id) AS buyers_today
    FROM buyer_purchases
    GROUP BY purchase_date
),
rolling_sets AS (
    -- Step 2: 使用窗口函数，收集过去 7 天（转为秒数滑动）的所有买家集合
    SELECT 
        purchase_date,
        -- 将过去 7 天的 Array 组合成一个大 Array
        flatten(
            collect_list(buyers_today) OVER (
                ORDER BY CAST(purchase_date AS DATE)
                RANGE BETWEEN INTERVAL 6 DAYS PRECEDING AND CURRENT ROW
            )
        ) AS rolling_buyers_array
    FROM daily_buyer_sets
)
-- Step 3: 对大 Array 进行去重计数
SELECT 
    purchase_date,
    size(array_distinct(rolling_buyers_array)) AS rolling_7d_active_buyers
FROM rolling_sets
ORDER BY purchase_date;

-- 错误示范！Spark SQL 不支持这种写法
SELECT purchase_date, 
       COUNT(DISTINCT buyer_id) OVER(ORDER BY CAST(purchase_date AS DATE) RANGE BETWEEN INTERVAL 6 DAYS PRECEDING AND CURRENT ROW)
FROM buyer_purchases;