Approach: compute per-user rolling 7-day aggregates that remain correct despite late events up to 48h by using event-time windows with watermarking, partitioned storage, and incremental materialized views (or streaming upserts). Use daily partitions for input, incremental processing that reprocesses the last 3 days (48h + slack) and merges into a materialized daily-per-user base; derive 7-day rolling metrics by joining the last 7 daily buckets.Example (Spark SQL / Databricks / BigQuery-style):sql
-- 1) ingest events into partitioned table by event_date (YYYY-MM-DD) and append
CREATE TABLE events (
user_id STRING,
event_time TIMESTAMP,
value DOUBLE
) PARTITIONED BY (event_date DATE);
-- 2) compute daily per-user aggregates (materialized daily view) with idempotent upsert
MERGE INTO daily_user AS D
USING (
SELECT user_id, DATE(event_time) AS event_date,
COUNT(*) AS cnt, SUM(value) AS sum_value
FROM events
WHERE event_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 5 DAY) -- process window
GROUP BY user_id, DATE(event_time)
) S
ON D.user_id = S.user_id AND D.event_date = S.event_date
WHEN MATCHED THEN UPDATE SET cnt = S.cnt, sum_value = S.sum_value
WHEN NOT MATCHED THEN INSERT (user_id, event_date, cnt, sum_value) VALUES (...)
;
-- 3) rolling 7-day metric from daily_user
SELECT u.user_id,
SUM(u.sum_value) OVER (PARTITION BY u.user_id ORDER BY u.event_date
ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) AS sum_7d,
SUM(u.cnt) OVER (PARTITION BY u.user_id ORDER BY u.event_date
ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) AS cnt_7d
FROM daily_user u
WHERE u.event_date BETWEEN DATE_SUB(CURRENT_DATE(), INTERVAL 6 DAY) AND CURRENT_DATE();
