Батч-пайплайны и оркестрация: ETL vs ELT, Airflow, dbt

На сайте есть событийные пайплайны и стрим-процессинг, но огромный пласт работы с данными по-прежнему батчевый: раз в сутки собрать, преобразовать, разложить по витринам. У батча свои вопросы. Не «как не потерять событие в потоке», а «что останется в хранилище, если прогон упал на середине и его перезапустили», «какой день на самом деле обработала задача» и «что будет с историей, когда бизнес передумает, как считать выручку».

Эта статья отвечает на них счётом. Генератор строит магазин за 14 дней и сам считает правильный ответ — выручку, число заказов и позиций по дням, — по своим данным, а не по базе. Пайплайн на Airflow 3, dbt и PostgreSQL прогоняется в разных постановках, и после каждой хранилище сверяется с эталоном: совпало ли, сколько строк лишних, сколько потеряно.

Первая статья серии «Data engineering на практике». Батч-двойник событийных пайплайнов; глубина самого Airflow — планировщик, исполнители, XCom — в отдельной серии про AirflowСкоро.

Путь данных серии и граница, которую разбирает каждая статья:

flowchart LR
    SRC["Источник
база заказов"] -->|"выгрузка дня
Airflow"| RAW["raw
сырые строки"]
    RAW -->|"проверки, карантин,
контракт"| STG["staging
dbt"]
    STG --> MART["витрины
факты и измерения"]
    RAW -.->|"озеро"| LAKE["таблица Iceberg
в S3"]
    P["Батч-пайплайны:
интервалы, повторы,
поздние данные"] -.- SRC
    Q["Качество и контракты:
что пускать дальше"] -.- STG
    M["Моделирование:
гранулярность, SCD, ключи"] -.- MART
    I["Форматы таблиц:
конкурентная запись,
удаление, история"] -.- LAKE
    O["Наблюдаемость:
мониторы и lineage
над всем путём"] -.- MART

В статье

• Стенд
• ETL против ELT: цена смены правила
• Идемпотентность: повтор и падение посреди загрузки
• Какой день берёт задача
• Backfill истории
• Поздние данные
• Повторы задач
• Как выбирать
• Чего эти результаты не значат

Стенд

Один PostgreSQL 18 играет две роли: база shop со схемой app — источник (заказы и позиции), база dwh — хранилище со слоями raw (сырые строки как в источнике), staging и marts (витрины dbt). Airflow 3.3 работает на LocalExecutor, dbt-core 1.12 с dbt-postgres 1.11 — в отдельном окружении, чтобы зависимости одного не меняли зависимости другого. Данные — 14 дней, 01–14.09.2026, около 300 заказов в день.

Главный DAG — elt_daily: выгрузить заказы дня из источника в raw, затем dbt run строит дневную выручку. Время Airflow стенд не ждёт: каждый запуск — явная логическая дата через airflow dags test, airflow dags trigger или airflow backfill create. История лежит в прошлом намеренно: backfill не создаёт запусков на даты позже текущей, и история «в будущем» дала бы разное число запусков в зависимости от дня прогона.

ETL против ELT: цена смены правила

ETL преобразует данные до загрузки, ELT загружает сырые строки как есть и преобразует их уже в хранилище — здесь dbt. Преобразование в ETL не обязательно агрегация: это может быть очистка и нормализация, после которых в хранилище приходят подробные факты. Поэтому сравнивать надо два вопроса раздельно — где идёт преобразование и что из исходной детализации сохранено. Разница видна не в день запуска, а в день, когда бизнес меняет правило.

Постановка. Три схемы. etl_daily — ETL с сохранением только агрегата: выручка считается в Python, в хранилище — итог дня. etl_facts_daily — ETL с подробными фактами: Python считает сумму каждого заказа, в хранилище — факт заказа со статусом, выручку по правилу считает SQL. elt_daily — ELT с сохранением raw: заказы и позиции в raw, выручку считает dbt. Все схемы прогоняются за 14 дней на правиле «выручка без отменённых» (контроль: совпадение с эталоном). Затем правило меняется — без возвратов тоже — и история пересчитывается так, как позволяет схема. Запросы к источнику считает pg_stat_statements базы источника, а не сам пайплайн.

Контроль сошёлся: на прежнем правиле все схемы совпадают с эталоном во все дни.

$ python3 scripts/run-elt.py etl
{"variant": "etl/initial", "outcome": "match", "detail": "совпало с эталоном v1 дней 14 из 14, запросов к источнику 28"}
{"variant": "etl/rule-change", "outcome": "match", "detail": "совпало с эталоном v2 дней 14 из 14, запросов к источнику 28"}
$ python3 scripts/run-elt.py etl-facts
{"variant": "etl-facts/initial", "outcome": "match", "detail": "совпало с эталоном v1 дней 14 из 14, запросов к источнику 28"}
{"variant": "etl-facts/rule-change", "outcome": "match", "detail": "совпало с эталоном v2 дней 14 из 14, запросов к источнику 0"}
$ python3 scripts/run-elt.py elt
{"variant": "elt/initial", "outcome": "match", "detail": "совпало с эталоном v1 дней 14 из 14, запросов к источнику 28"}
{"variant": "elt/rule-change", "outcome": "match", "detail": "совпало с эталоном v2 дней 14 из 14, запросов к источнику 0"}

ETL с сохранением только агрегата выбросил статус заказа на этапе преобразования, и пересчитать историю без источника нельзя: 14 повторных выгрузок, по два запроса на день. ETL с подробными фактами тоже преобразует до загрузки, но статус и сумма каждого заказа остались в хранилище — новое правило пересчитывается SQL без единого запроса к источнику. ELT с сохранением raw делает то же самое новым SQL в dbt поверх сырых строк. Дело, выходит, не в том, где идёт преобразование, а в том, что из детализации сохранено: факт заказа переживает смену правила о статусах, но не переживёт правило, которому нужно поле, отброшенное при очистке, — а сырые строки переживут. Здесь источник — свой PostgreSQL; в жизни это может быть чужая система с лимитом запросов или данные, которых там уже нет. Цена ELT — место под сырые строки и то, что они лежат в хранилище со всеми полями, включая те, которые лучше бы не копировать.

Идемпотентность: повтор и падение посреди загрузки

Повторный запуск — обычное дело: упал соседний шаг, поправили код, пересчитываем день. Вопрос — что повтор делает с уже загруженным.

Постановка. День 03.09 (686 позиций) загружается в raw четырьмя способами:

# pipeline/load.py
        if mode in ("delete-insert", "delete-batch"):
            cur.execute("DELETE FROM raw.order_items WHERE load_day = %s", (day,))
            cur.execute("DELETE FROM raw.orders WHERE load_day = %s", (day,))
            if mode == "delete-batch":
                c.commit()

append — просто вставка пакетами по 50 строк с фиксацией после каждого; delete-insert — удаление дня и вставка в одной транзакции; delete-batch — удаление фиксируется сразу, вставка идёт пакетами; merge — вставка с ON CONFLICT (id) DO UPDATE пакетами. Сценарии: один прогон (контроль); повтор того же дня; день уже загружен, повторная загрузка падает после пяти пакетов позиций — счёт сразу после падения и после ещё одного повтора.

Контроль сошёлся: один прогон без падения во всех режимах даёт эталон.

$ python3 scripts/run-idempotency.py append crash-reload
{"variant": "append/crash-reload", "outcome": "mismatch", "detail": "итог: дублей 936, потерь 0; после падения: строк 936, дублей 250, потерь 0"}
$ python3 scripts/run-idempotency.py delete-batch crash-reload
{"variant": "delete-batch/crash-reload", "outcome": "match", "detail": "итог: дублей 0, потерь 0; после падения: строк 250, дублей 0, потерь 436"}

append неидемпотентен в самом прямом смысле: каждый повтор добавляет день целиком, а падение оставляет половину. Остальные три режима сходятся к эталону после повтора — разница в том, что видит читатель хранилища между падением и повтором. Транзакция откатывается целиком, и день остаётся прежним. delete-batch успел удалить старое и вставить 250 строк из 686: если в этот момент витрина пересчитается или аналитик откроет отчёт, выручка за день будет почти втрое меньше настоящей, и ничто не скажет, что день неполный. merge на повторной загрузке оставил день целым: старые строки на месте, часть успела обновиться. На первой загрузке дня такого запаса у него нет.

Отсюда практическое правило: повтор должен заменять, а не добавлять, а замена — быть атомарной для читателя. В PostgreSQL это одна транзакция; в хранилищах без многооператорных транзакций — загрузка в новую партицию и подмена указателя на неё, как это делают табличные форматы, о которых третья статья серииготовится, с 2 ноября.

Какой день берёт задача

Батчевая задача почти всегда работает «за период»: выгрузить заказы за сутки. Откуда она узнаёт, за какие сутки, — самый частый источник тихих ошибок.

Постановка. DAG interval_demo запускается за 14 логических дат, как при пересчёте истории. В одном варианте задача берёт день из data_interval_start запуска, в другом — «вчера» по часам машины (now() - 1 day, так часто пишут в cron). «Сегодня» для опыта зафиксировано — 15.09, следующий день после истории: иначе ответ зависел бы от даты прогона стенда.

Задача по часам при пересчёте истории 14 раз обработала один и тот же день и ни разу — остальные 13. В ежедневной работе она выглядит правильной, ломается ровно тогда, когда запуск идёт не в свой день: backfill, повтор после выходных, догоняющий запуск после простоя.

Но и data_interval_start в Airflow 3 значит не то, к чему привыкли по Airflow 2. Постановка. Два DAG с @daily: с расписанием по умолчанию и с CronDataIntervalTimetable. Снятие паузы создаёт обычный запуск за последний срок; задача пишет, какой день ей достался относительно дня срабатывания. Отдельно — ручной запуск airflow dags trigger --logical-date 2026-09-03.

# airflow/dags/sched_dags.py
make("sched_trigger", "@daily")
make("sched_interval", CronDataIntervalTimetable("@daily", timezone="UTC"))

$ python3 scripts/run-interval.py manual-trigger
{"variant": "manual-trigger", "outcome": "mismatch", "detail": "по data_interval_start взят день запуска, а не логической даты"}

В Airflow 3 @daily по умолчанию — расписание без интервала данных (create_cron_data_intervals = False): запуск в полночь получает начало и конец интервала, равные моменту срабатывания. Задача, которая выгружает заказы с data_interval_start до data_interval_end, выгрузит пустой промежуток, а задача, которая берёт день data_interval_start, — сутки, которые только начались. В Airflow 2 тот же DAG получал прошедшие сутки. Вернуть это поведение можно расписанием CronDataIntervalTimetable или настройкой [scheduler] create_cron_data_intervals = True (расписания Airflow 3). Замена data_interval_start на logical_date здесь не поможет: у такого расписания логическая дата — тоже момент срабатывания, и задача по-прежнему возьмёт начавшийся день. Без интервального расписания прошедшие сутки надо вычислять в задаче явно — например, как день перед логической датой.

Ручной запуск с логической датой — вторая неожиданность. airflow dags trigger --logical-date 2026-09-03 создал запуск с логической датой 03.09, но с data_interval_start в момент запуска. Задача по интервалу выгрузила «сегодня», где данных нет, и завершилась успешно — стенд заметил это только потому, что сверял хранилище с эталоном. airflow dags test с той же датой даёт интервал по логической дате, поэтому на тестах этого не видно. Задачам, которые запускают вручную за прошлое, надёжнее брать день из logical_date.

Backfill истории

Пересчитать историю — обычная операция: новый столбец в витрине, исправленная ошибка, новый пайплайн поверх старых данных. В Airflow 3 для этого есть airflow backfill create, и запуски backfill выполняет планировщик.

Постановка. DAG bf_free и bf_ordered делают одно и то же — выгрузку дня (delete-insert) и витрину dbt, — второй с depends_on_past=True. Backfill за 14 дней; повтор того же backfill; сбой выгрузки за 05.09.

Контроль сошёлся: полный backfill — 14 запусков, витрина совпадает с эталоном.

Повтор того же backfill по умолчанию ничего не делает: даты уже обработаны. Пересчитать их можно только явно — --reprocess-behavior completed создаёт запуски заново, а идемпотентная загрузка даёт тот же эталон. Сбой одного дня без depends_on_past не мешает остальным: 13 дней посчитаны, один — нет, и витрина неполна ровно на этот день. С depends_on_past всё, что после сбоя, стоит: 4 дня до сбоя, сбой, 9 невыполненных. Это и цель, и цена: следующий день не построится на дыре, но и не построится вовсе, пока сбой не исправят.

Для таких DAG Airflow 3 ещё и строже с самим backfill. Первый запуск — 03.09 уже обработан ручным запуском, второй — диапазон чистый:

$ python3 scripts/run-backfill.py ordered-default
{"variant": "ordered-default", "outcome": "mismatch", "detail": "backfill отказан: airflow.models.backfill.InvalidReprocessBehavior: Dag has tasks for which depends_on_past=True. You must set reprocess behavior to reprocess completed or reprocess failed."}
$ python3 scripts/run-backfill.py ordered-clean
{"variant": "ordered-clean", "outcome": "mismatch", "detail": "backfill отказан: airflow.models.backfill.InvalidReprocessBehavior: Dag has tasks for which depends_on_past=True. You must set reprocess behavior to reprocess completed or reprocess failed."}

Backfill DAG с depends_on_past без --reprocess-behavior (или с none) отказан всегда — даже на диапазоне, где ни одна дата не обработана: проверка в Airflow 3.3 смотрит только на флаг задачи и значение параметра (_validate_backfill_params), а не на историю запусков. Выбор поведения для обработанных и упавших дат Airflow требует сделать явно. --run-backwards для таких DAG отказан всегда. Исполнителю для всего этого нужен API-сервер Airflow: задачи получают работу через Execution API, и без него запуски висят в очереди — airflow dags test этого не показывает, он выполняет задачи в одном процессе.

Поздние данные

Не всё, что произошло 05.09, приходит в источник 05.09: мобильный клиент без сети, партнёр с ночной выгрузкой, ретраи в чужой системе.

Постановка. Дни 01–06.09 обработаны; затем в источник приходят 20 заказов за последний час 05.09 — с опозданием в 2 дня; дальше обрабатываются 07–14.09. Выгрузка — по времени события (created_at) или по времени поступления (ingested_at), с окном пересмотра прошлых дней и без; витрина — таблица, пересчитываемая целиком, или инкрементальная модель dbt:

-- dbt/models/marts/daily_revenue_inc.sql
{% if is_incremental() %}
where o.order_day >= (select max(order_day) from {{ this }}) - {{ var('lookback_days', 0) }}
{% endif %}

Контроль сошёлся: без поздних заказов — эталон.

$ python3 scripts/run-late.py event-lookback1
{"variant": "event-lookback1", "outcome": "mismatch", "detail": "за 2026-09-05 не хватает заказов 20 (выручки 200000); совпало дней 13 из 14"}
$ python3 scripts/run-late.py ingest-inc
{"variant": "ingest-inc", "outcome": "mismatch", "detail": "за 2026-09-05 не хватает заказов 20 (выручки 200000); совпало дней 13 из 14"}

Выгрузка по времени события забирает день один раз и больше к нему не возвращается: 20 заказов потеряны. Окно пересмотра помогает, только если оно не короче опоздания — окна в 1 день при опоздании в 2 мало. Выгрузка по времени поступления не теряет ничего на уровне raw: поздние строки приходят в день поступления. Но дальше всё решает витрина. Пересчитанная целиком, она раскладывает заказы по дате события и совпадает с эталоном. Инкрементальная пересчитывает только «свежие» дни и о 05.09 больше не вспоминает — те же 20 заказов потеряны уже на этом шаге. Окно пересмотра витрины или периодический --full-refresh это лечат.

Общее правило: окно пересмотра — это обещание, сколько опоздания пайплайн переносит. Его стоит выбирать по наблюдаемому опозданию источника и проверять мониторингом — об этом последняя статья серииготовится, с 5 ноября.

Повторы задач

retries в Airflow — первая защита от переходных ошибок: сеть моргнула, база перегружена. Вопрос — что повтор делает с тем, что упавшая попытка уже записала.

Постановка. DAG retry_r1 и retry_r3 (retries=1 и 3) грузят 03.09; задача падает на первых попытках после записи пяти пакетов позиций. Запуск — airflow dags trigger, повторы выполняет планировщик; число попыток — из task_instance.try_number.

Контроль сошёлся: задача без сбоев — успех с первой попытки, эталон.

$ python3 scripts/run-retries.py r3-append-k1
{"variant": "r3-append-k1", "outcome": "mismatch", "detail": "задача success на попытке 2 (retries=3, падений 1); дублей 250, потерь 0"}

Повторы спасают задачу, пока сбоев подряд не больше, чем retries, — и только если задача идемпотентна. Неидемпотентная загрузка после успешного повтора выглядит успешной: задача зелёная, а в raw 250 лишних строк от упавшей попытки. retries повторяет задачу, а не откатывает её.

Как выбирать

Чего эти результаты не значат

• Ничего о скорости и масштабе. 14 дней по ~300 заказов в одном PostgreSQL; длительность прогонов стенд не публикует.
• Версии: Airflow 3.3.2, dbt-core 1.12.5, dbt-postgres 1.11.0, PostgreSQL 18.6. Интервалы, ручной запуск и backfill — поведение Airflow 3; в Airflow 2 оно другое.
• Время управляемое. Опыты не ждут реального расписания: явные логические даты, dags test, dags trigger, backfill create. Обычный запуск планировщика — только в опыте с расписаниями; его дата зависит от дня прогона, поэтому публикуется сдвиг, а не дата.
• Падение — искусственное: исключение после заданного числа пакетов, а не обрыв соединения или убитый процесс. Транзакция PostgreSQL откатывается и в тех случаях, но порядок фиксаций у реального загрузчика может быть другим.

Ссылки

• Стенд: digital-cookbook/data-engineering/pipelines — шесть осей, генератор с эталоном.
• Airflow: расписания и timetables
• Airflow: backfill
• dbt: инкрементальные модели
• Распределённый cron и чем он отличается от оркестратора — статья про планировщики; очереди фоновых задач — статья про очереди задач.