# Дневной конвейер трансформаций

Даг `etl_pipeline` собирает модель DDS, обновляет витрины DM и проверяет
результат. Здесь описаны запуск, выбор дней и запись результата в таблицы.
Состав моделей — в [справочниках DDS](dds/README.md) и [DM](dm.md).

## Форма

`etl_pipeline` последовательно запускает `dds_transform`, `dm_transform`
и `dq_check`. Каждый следующий даг начинается после успеха предыдущего.
Для этого `TriggerDagRunOperator` ждет завершения запущенного им дага.
Если тот завершился ошибкой, задача ожидания в конвейере тоже завершится
ошибкой.
Поведение проверено через Context7 по документации Airflow 3.x (3.1.6).

У этих дагов нет расписания. Их вызывает конвейер; для отдельного разбора
можно запустить даг слоя вручную. У каждого разрешен один активный запуск:
два запуска одного дага используют одни промежуточные таблицы и могли бы
стереть результаты друг друга. Расчет каждой модели выполняется отдельной
задачей или группой задач.

## Три пути запуска

```text
world_initialize ─┐
world_live_day ───┼─> etl_pipeline ─> dds_transform ─> dm_transform ─> dq_check
запуск рукой ─────┘

world_next_day ─────> источники без ETL
```

- `world_live_day` запускает конвейер после успешной генерации, приема
  заказов и обновления `world_position`. Если трансформация упала, день
  уже прожит и позиция не откатывается. После исправления причины
  запустите `etl_pipeline` вручную.
- `world_next_day` отправляет дни пачкой и ждет приема заказов, но конвейер
  не запускает. После него запустите `etl_pipeline` сами.
- `world_initialize` запускает конвейер последним шагом: при успешном
  первом `make up` начальные данные уже собраны в DDS и DM.

## Объём прогона

Объем прогона — дни, которые обрабатывает один запуск. Даты передавать
не нужно: даги определяют их по состоянию таблиц и `world_position`.
Например, после генерации семи дней без ETL один запуск рассчитает сессии
за все семь дней.

Для расчета сессий выбираются дни от первого, за который в DDS еще нет
сессий, до последнего прожитого.
`world_position` хранит номер следующего дня для генерации, поэтому
последний прожитый день равен `min(EventDate) + world_position - 1`.
Начало календаря берется из ODS; события ранних дней там не удаляются по TTL.
Один `max(EventDate)` не годится: события за день есть и во время его
генерации, когда последние визиты еще не закончились.
Точный запрос — [`session_scope.sql`](../../sql/dds/session_scope.sql).

Для заказов диапазон задает последний принятый слепок ODS. Среди строк
с максимальной `snapshot_date` выбираются самая ранняя и самая поздняя
даты оформления заказа. Состояние заказов за эти дни еще может измениться.
Если в ODS есть заказы за более ранний день, а в DDS за него нет строк,
начало диапазона сдвигается до этого дня. Запрос заново рассчитывает
заказы за все дни выбранного отрезка.

Позиция мира для выбора дней заказов не подходит: с событиями дня D
приходит слепок за D−1. Объем рассчитывается по принятым строкам ODS;
их наличие еще не доказывает полноту слепка.

Карта идентичностей пересчитывается целиком. Витрины выбирают дни
по своим источникам — см. [ритм пересборки DM](dm.md#ритм-пересборки).
Если за день сохранилась хотя бы одна строка, запрос не считает его
пропущенным. Полноту данных проверяют отдельно.

## Готовность источника

Перед сборкой DDS задача `wait_for_source` последовательно ждет двух
условий:

1. Группа `clickstream_hits` дочитала оба раздела топика `hits`.
   Подтвержденные смещения сравниваются с верхними смещениями Kafka.
   Группа и смещения должны существовать, суммарное отставание — равняться нулю.
2. В `system.distribution_queue` у `ods.event_dist` не осталось файлов
   доставки на шарды. Запрос через `clusterAllReplicas` читает обе ноды.

Опросы идут с паузой в три секунды; на обе стадии задан общий срок
300 секунд. Журнал показывает смещения и отставание Kafka, затем число
файлов и `last_exception` очереди на каждой ноде. Если готовность не
подтверждена за срок ожидания, задача завершается ошибкой.

Одного нулевого отставания Kafka мало: сообщение может быть подтверждено,
пока Distributed-таблица еще доставляет его на шард. Успех `orders_ingest`
тоже не заменяет эти проверки: у заказов другой топик и потребитель.

Эти условия проверяют доставку на момент опроса. Они не останавливают
поступление новых событий и не сверяют содержимое таблиц. Поэтому
`world_position` отдельно ограничивает сборку сессий прожитыми днями,
а качество результата проверяет `dq_check`.

## Идемпотентность

Идемпотентность означает, что повторная обработка тех же исходных данных
не добавляет копии деловых строк. Служебные `_load_id` и `_load_ts`
при новой сборке меняются.

Результат дневного расчета сначала записывается в промежуточную таблицу.
Она нужна, чтобы читатели продолжали видеть прежние данные, пока новые
еще рассчитываются. Например, для сессий это `dds.session_stage_rep`,
а основная таблица — `dds.session_rep`.

Задачи выполняются в таком порядке:

1. `scope` выбирает даты для расчета.
2. `rebuild` очищает промежуточную таблицу, рассчитывает строки за выбранные
   дни и записывает их через `_stage_dist`. Настройка
   `distributed_foreground_insert = 1` заставляет вставку дождаться
   доставки строк на шарды.
3. `replace` заменяет в основной таблице раздел за каждый выбранный день
   готовым разделом промежуточной таблицы. Раздел таблицы называется
   партицией; для замены используется `REPLACE PARTITION`.

Структура и распределение строк по шардам у двух таблиц должны совпадать.
Замена выполняется на локальных `_rep` с `ON CLUSTER`. На каждом шарде
читатель видит либо старый, либо новый раздел целиком — это атомарность
замены. Общей транзакции для двух шардов и всех дней нет: при сбое один
шард может уже содержать новый результат, а другой — прежний. Повторить
`replace` можно, пока промежуточная таблица хранит тот же результат;
копии строк в основной таблице от этого не появятся.

Новый запуск всего конвейера заново выбирает дни. Он не обязан повторить
все дни неудачного запуска: если дата уже есть в целевой таблице, поиск
пропусков ее не выберет. В частности, изменение SQL сессий само по себе
не заставляет пересчитать заполненные дни. Повторная сборка выбранного
дня разобрана в [задании ступени 4](../labs/4-1-session-pipeline.md).

Карта идентичностей и `daily_traffic` обновляются иначе: каждый запуск
добавляет полный результат новой версией в `ReplacingMergeTree(_load_ts)`.
Публичные представления читают текущие строки через `FINAL`. Старые версии
могут физически оставаться до фонового слияния; повторяемость относится
к результату чтения, а не к числу строк на диске.

## Чего конвейер о себе не знает

Трансформация собирает слой по правилу, а не сверяет его с источником. Отсюда
три вещи, которых она не заметит, и у всех трёх один адрес — контур DQ.

- **Часть последнего слепка не доехала.** Отказ вставки после коммита офсетов
  уносит прочитанное с собой. Пропади так строки самого старого дня окна —
  окно, выведенное из ярлыков, молча сузится, и день не пересоберётся
  ([`sql/dds/order_scope.sql`](../../sql/dds/order_scope.sql)).
- **Замена партиции прошла на одном шарде из двух.** День в слое есть — значит,
  объём следующего прогона его не тронет, а слой останется собранным наполовину
  из старого, наполовину из нового.
- **Брак внутри `items` становится нулями.** Строгий приём внутрь позиций не
  смотрит, разбор по типу молчит, и цена без числа доедет до витрины выручки
  нулём ([`sql/dds/order_rebuild.sql`](../../sql/dds/order_rebuild.sql)).

Общее у них одно, и оно же подсказывает лекарство: ответить может только
сверка слоя с источником по дню — счёт заказов и позднейший `updated_at` с
обеих сторон. Трансформации такую сверку не поручают намеренно: она собирает,
а проверяет другой. Так две роли остаются различимы, и ни одна не
притворяется другой.

## Не входит

- Партиционная переобработка — переиграть произвольный прошлый день или
  диапазон, включая форму необязательных параметров того же дага.
- Инкрементальный ETL вместо пересборки отклонен (#2, wontfix). Идея отдельной
  лабы на эту тему — #149.
- Состав витрин `dm_transform` и проверок `dq_check` —
  [дока слоя DM](dm.md).

## Отклонённые варианты

- **Задачи трансформаций прямо в шаговых дагах.** Один зелёный прогон —
  весь день, но даги времени отвечают за ось времени: шаговых дагов два, шаги
  дублировались бы, а этапу 5 пришлось бы выдирать их обратно.
- **Make-цель или SQL рукой при приёмке.** Второй оркестратор рядом с
  Airflow — против [ADR 0013](../adr/0013-world-lifecycle.md), — а в живом
  режиме слои копили бы отставание молча.
- **Временный даг с нейтральным именем до этапа 5.** Дневная сборка вперёд
  никуда не денется, переезд был бы переименованием ради выдуманной границы.
- **Диапазон дней параметрами триггера.** Ручному запуску менти параметры
  передавать некому, а перезапуск после падения требовал бы помнить conf
  упавшего прогона.
- **Принять гонку с хвостом `hits`.** Дыра всплыла бы в DQ-сверке, а
  чиниться — не раньше этапа 5.

## Связанные решения

- [ADR 0003](../adr/0003-dag-run-state.md) — зелёный конец графа не
  переживает отказ выше: отсюда ожидания в триггерах.
- [ADR 0009](../adr/0009-world-control.md) — даги модельного времени: их шаговый
  даг живого дня и дёргает конвейер.
- [ADR 0013](../adr/0013-world-lifecycle.md) — жизненный цикл мира: почему
  прикладным процессом владеет Airflow и почему `world_initialize` кончается
  собранными слоями.
- Резолюция — тикет #134 карты #127; что пересобирает каждая сущность —
  [заказ](dds/order.md), [сессия](dds/session.md),
  [карта идентичностей](dds/identity.md).