# Дневной конвейер трансформаций Как принятые источники превращаются в модель DDS и витрины DM: кто запускает трансформации, когда и в каком объёме. Правила самих сущностей — зерно, движки, что именно пересобирается — живут в доках слоя ([карта модели DDS](dds/README.md)); здесь — только конвейер. Спека обещала `etl_pipeline` этапом 5 — он понадобился раньше. Конвейер рождается этапом 4, и временного в нём ничего нет: дневная сборка вперёд останется и после. Этапу 5 достаётся его законный предмет — партиционная переобработка, умение переиграть произвольный прошлый день по команде. ## Форма Дневной конвейер на стенде один — даг `etl_pipeline`. Своей работы у него нет: он дёргает слойные даги последовательно и ждёт каждый — `dds_transform`, затем `dm_transform`. Контур DQ этапа 4 встаёт его же звеном после DM; состав звена — в [доке слоя DM](dm.md). Слойные даги устроены по образцу `orders_ingest`: без расписания, один активный прогон, запускаются триггером конвейера или рукой при разборе происшествий. Витрины внутри `dm_transform` — отдельные задачи. ## Три пути запуска - **Живой день.** `world_live_day` дёргает конвейер с ожиданием — последним шагом, после установки позиции. Порядок несёт смысл: день сыгран — факт, упавшая трансформация его не отменяет, поэтому позиция ставится раньше; но зеленеть, не узнав, разложился ли день, работник не должен ([ADR 0003](../adr/0003-dag-run-state.md)). Такой отказ чинится перезапуском `etl_pipeline` рукой — перезапуску ничего не нужно помнить, см. «Объём прогона». - **Разгон пачкой.** `world_next_day` конвейер не дёргает: пульт управляет генератором и только им. После разгона менти запускает конвейер сам — и видит, как слои догоняют уехавший вперёд мир. - **Рождение мира.** `world_initialize` дёргает конвейер последним шагом: `make up` отдаёт стенд с собранными слоями, а не пустые витрины. Уроки ручного запуска начинаются со следующих дней. ## Объём прогона Параметров у конвейера нет: объём он выводит сам из `world_position` и состояния слоёв. Сессии догоняют хвост — от первого несобранного дня по последний прожитый; заказы пересобирают дышащее окно K и тот же хвост; карта идентичностей пересчитывается целиком; правила витрин — у [слоя DM](dm.md). Поэтому один и тот же запуск одинаково верен из триггера, с кнопки менти и после падения, а разгону на семь дней хватает одного прогона. Окно заказов отсчитывается от **последнего дня, о котором рассказал сам источник**, — от позднейшего дня заказа в ODS. Рядом напрашиваются ещё два способа, и оба отвергнуты по одной причине: они отвечают не на тот вопрос. Позиция мира говорит, какой день прожит, а не какой доехал: прогон дня D везёт слепок дня D − 1, да и позиция ставится раньше трансформации, так что после неудачного приёма она обгоняет хранилище. Слой обязан считать объём по тому, что у него есть, а не по тому, что мир о себе заявляет. У сессий всё наоборот: им отвечает только позиция, потому что вопрос у них другой. Им нужен не день, о котором рассказал источник, а день, который мир **прожил**: события живой день от прожитого не отличают — и посреди дня D, и сразу после его конца позднейший день событий равен D. Соберись сессии посреди дня, визит, начатый в 23:50, остался бы обрезанным навсегда: замёрзший день слой не пересобирает. Отсюда правило — последний прожитый день равен первому дню событий плюс позиция минус один ([`sql/dds/session_scope.sql`](../../sql/dds/session_scope.sql)). Оговорка про «доехал» остаётся в силе и здесь: за хвост потока отвечает не позиция, а «Готовность источника». Ярлыки доставки — `snapshot_date`, `_load_id`, `_load_ts` — не отвечают тоже, и это свойство ODS, а не недосмотр: слой схлопывает доставки, сохраняя деловые версии, поэтому у заказа, который не менялся, ярлык выжившей строки достаётся жребием ([приём заказов](orders/ingestion.md), «Роль ODS»). Замер на стартовом мире: последний слепок нёс все 1662 заказа, а его ярлык сохранили 1188 — отбор «дни, у которых `snapshot_date` равен последнему» потерял бы четверть заказов и молча пропустил бы дальний край окна, где заказы давно замёрзли. К времени создания это не относится: у всех версий заказа оно одно. ## Готовность источника Первым шагом `dds_transform` обязан дождаться, пока хвост топика `hits` дожёван: отставание его потребителей нулевое (`system.kafka_consumers`). Зелёный приём заказов о событиях не говорит ничего — топики разные и потребители разные. В живом режиме зазор ничтожен, но в пакетном генератор выстреливает дни в топик быстрее, чем чтец их прожёвывает, и сессии собираются по недожёванному дню. DQ-сверка такой день подсветит, но починить его до этапа 5 нечем: профилактика дешевле ремонта, которого ещё нет. Шага этого в даге ещё нет — он приезжает отдельным тикетом (#156), и до тех пор дыра открыта: сессии считают день прожитым по позиции мира, а хвост потока не стережёт никто. На приёмке отставание смотрят руками. ## Идемпотентность Партиция не заменяется простой вставкой. День собирается в таблицу-двойник — вставка через Distributed раскладывает строки по шардам, — затем `REPLACE PARTITION` на кластере атомарно подменяет кусок каждого шарда. Эта механика — часть самих трансформаций и строится этапом 4: без неё дневной прогон не повторяем. Сущности, пересчитываемые целиком, идемпотентны и без замены: полная перезапись в `ReplacingMergeTree` — так живут карта идентичностей и витрина `daily_traffic` ([слой DM](dm.md)). ## Чего конвейер о себе не знает Трансформация собирает слой по правилу, а не сверяет его с источником. Отсюда четыре вещи, которых она не заметит, и у всех четырёх один адрес — контур DQ. - **Окно K разъехалось с источником.** Число живёт двумя копиями: у генератора в `world.py` и у стенда в `compose.yaml`. Разойдутся — слой начнёт недобирать край окна или переписывать замёрзший день, и оба прогона будут зелёными. - **Пустой свежий день сдвигает край окна.** Если в позднейший день источника не родилось ни одного заказа, окно съезжает на день назад ([`sql/dds/order_scope.sql`](../../sql/dds/order_scope.sql)). - **Замена партиции прошла на одном шарде из двух.** День в слое есть — значит, объём следующего прогона его не тронет, а слой останется собранным наполовину из старого, наполовину из нового. - **Брак внутри `items` становится нулями.** Строгий приём внутрь позиций не смотрит, разбор по типу молчит, и цена без числа доедет до витрины выручки нулём ([`sql/dds/order_rebuild.sql`](../../sql/dds/order_rebuild.sql)). Общее у них одно, и оно же подсказывает лекарство: ответить может только сверка слоя с источником по дню — счёт заказов и позднейший `updated_at` с обеих сторон. Трансформации такую сверку не поручают намеренно: она собирает, а проверяет другой. Так две роли остаются различимы, и ни одна не притворяется другой. ## Не входит - Партиционная переобработка — переиграть произвольный прошлый день или диапазон, включая форму необязательных параметров того же дага, — этап 5 (#7); туда же стык #111 про дату модельного мира и `logical_date`. - Инкрементальный ETL вместо пересборки — техдолг #2. - Состав витрин `dm_transform` и звена DQ — [дока слоя DM](dm.md); имена задач — нарезка этапа 4. ## Отклонённые варианты - **Задачи трансформаций прямо в работниках пульта.** Один зелёный прогон — весь день, но пульт — про ось времени: работников два, шаги дублировались бы, а этапу 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).