В пайплайн ежедневной агрегации иногда приходят события с опозданием (late-arriving data) — например, событие случилось вчера, но долетело до хранилища только сегодня. Как спроектировать обработку, чтобы метрики за вчера не оставались навсегда заниженными?

Короткий ответ

Нужен механизм переобработки (reprocessing) за скользящее окно последних N дней при каждом запуске, либо отдельный механизм докладки опоздавших событий с пересчётом уже посчитанных агрегатов за затронутый период.

Как это работает подробнее

Вместо того чтобы агрегировать только «вчера» и считать период закрытым, пайплайн при каждом запуске пересчитывает, например, последние 3–7 дней — опоздавшие события, попавшие в raw-слой, учитываются при следующем пересчёте, а агрегат за прошлые дни обновляется.

Пример кода

example.sqlКопировать

-- Пересчёт агрегата за скользящее окно 7 дней при каждом запуске
DELETE FROM mart.daily_dau WHERE event_date >= CURRENT_DATE - 7;
INSERT INTO mart.daily_dau (event_date, dau)
SELECT event_date, COUNT(DISTINCT user_id)
FROM events_raw
WHERE event_date >= CURRENT_DATE - 7
GROUP BY event_date;

Что использовать на практике

• На практике. Размер окна пересмотра выбирают по эмпирическому распределению задержки событий (например, 95% событий долетают в течение 24 часов, но на окно всё равно берут с запасом 3–7 дней); также полезно иметь метрику «доля событий, пришедших с опозданием более N часов» для мониторинга.
• Проверка. сравнить значение dau за одну и ту же дату в двух последовательных запусках пайплайна (например, через день) — рост значения указывает на то, что докладка и пересчёт действительно работают.

Подводные камни

• Частые ошибки. Считают данные за день «окончательными» сразу после полуночи и больше никогда не пересчитывают — в этом случае метрики за прошлые дни систематически занижены, и это особенно заметно в финансовой отчётности и ретеншн-метриках; либо делают окно пересчёта слишком большим, что резко увеличивает стоимость и время пайплайна.