Для критичного ежедневного Airflow-пайплайна, который считает финансовые метрики, нужно спроектировать обработку бэкфиллов (пересчёт за прошлые даты) и защиту от задвоения данных при повторных запусках. Какие подходы вы рассмотрите?

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

Единственно верного решения нет: ключевая идея — сделать задачи идемпотентными (повторный запуск за ту же дату не создаёт дублей, а замещает результат), использовать execution_date/data_interval для параметризации запросов по датам и явно продумать стратегию бэкфилла (полный пересчёт диапазона дат vs только изменившихся партиций) с учётом объёма данных и критичности метрик.

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

• Как это работает: идемпотентность обычно достигается через паттерн DELETE + INSERT или MERGE/UPSERT по ключу даты и измерений вместо простого INSERT, так что повторный запуск задачи за ту же дату перезаписывает, а не добавляет данные; Airflow передаёт в задачу логический интервал выполнения (data_interval_start/end), который используется для фильтрации обрабатываемых данных, а не момент фактического запуска.

Пример кода

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

-- Идемпотентная загрузка за конкретный интервал (BigQuery)
DELETE FROM finance.daily_revenue
WHERE report_date = DATE('{{ data_interval_start }}');
INSERT INTO finance.daily_revenue
SELECT report_date, SUM(amount) AS revenue
FROM raw.transactions
WHERE DATE(created_at) = DATE('{{ data_interval_start }}')
GROUP BY report_date;

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

• На практике: для бэкфилла нужно решить, можно ли позволить себе пересчитать весь исторический диапазон (просто и надёжно, но дорого по времени/ресурсам для больших объёмов) или нужно считать только затронутые партиции (быстрее, но требует точно знать, что именно изменилось в источнике); также важно учитывать зависимые даунстрим-таблицы — пересчёт исторических данных в одной модели может потребовать каскадного пересчёта зависимых витрин.
• Пример и проверка. Проверка: запустить задачу дважды подряд за одну и ту же дату (airflow tasks test ... 2024-05-01 дважды) и убедиться, что SELECT COUNT(*) FROM finance.daily_revenue WHERE report_date = '2024-05-01' возвращает одну строку, а не две.

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

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