clickstream-data-platform/docs/architecture/etl.md

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

Как принятые источники превращаются в модель DDS и витрины DM: кто запускает трансформации, когда и в каком объёме. Правила самих сущностей — зерно, движки, что именно пересобирается — живут в доках слоя (карта модели DDS); здесь — только конвейер.

Дневная сборка вперёд появляется на этапе 4 и остаётся основным режимом конвейера. На этапе 5 к ней добавляется партиционная переобработка: запуск произвольного прошлого дня по команде.

Форма

Дневной конвейер на стенде один — даг etl_pipeline. Своей работы у него нет: он дёргает слойные даги последовательно и ждёт каждый — dds_transform, затем dm_transform и dq_check. Состав проверок описан в доке слоя DM.

Ждущие вызовы сделаны TriggerDagRunOperator. Проверка через Context7 по ближайшей доступной документации Airflow 3.x (версия 3.1.6) подтвердила, что оператор опрашивает точный созданный им прогон и краснеет вслед за потомком.

Слойные даги устроены по образцу orders_ingest: без расписания, один активный прогон, запускаются триггером конвейера или рукой при разборе происшествий. Витрины внутри dm_transform — отдельные задачи.

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

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

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

• Живой день. world_live_day дёргает конвейер с ожиданием — последним шагом, после установки позиции. Порядок несёт смысл: день сыгран — факт, упавшая трансформация его не отменяет, поэтому позиция ставится раньше; но зеленеть, не узнав, разложился ли день, работник не должен (ADR 0003). Такой отказ чинится перезапуском etl_pipeline рукой — перезапуску ничего не нужно помнить, см. «Объём прогона».
• Разгон пачкой. world_next_day конвейер не дёргает: пульт управляет генератором и только им. После разгона менти запускает конвейер сам — и видит, как слои догоняют уехавший вперёд мир.
• Рождение мира. world_initialize дёргает конвейер последним шагом: make up отдаёт стенд с собранными слоями, а не пустые витрины. Уроки ручного запуска начинаются со следующих дней.

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

Параметров у конвейера нет: объём он выводит сам из world_position и состояния слоёв. Сессии догоняют хвост — от первого несобранного дня по последний прожитый; заказы пересобирают дышащее окно и тот же хвост; карта идентичностей пересчитывается целиком; правила витрин — у слоя DM. Поэтому один и тот же запуск одинаково верен из триггера, с кнопки менти и после падения, а разгону на семь дней хватает одного прогона.

Окно заказов задаёт последний слепок, наблюдаемый в ODS. У каждого заказа snapshot_date равен последнему слепку, в котором источник его показал. Трансформация выбирает строки с максимальной snapshot_date и берёт среди них минимальный и максимальный order_date. Это границы дней, состояние которых источник ещё может изменить (ADR 0014). Кроме живого окна, трансформация закрывает пропуски: находит самый ранний день, представленный в ODS и ещё не собранный в DDS, затем при необходимости расширяет до него левую границу.

Позиция мира заказам не отвечает: она говорит, какой день прожит, а не какой доехал. Прогон дня D везёт слепок дня D − 1, да и ставится позиция раньше трансформации, так что после неудачного приёма обгоняет хранилище. Слой обязан считать объём по тому, что у него есть, а не по тому, что мир о себе заявляет.

Для сессий границу задаёт world_position: она отличает полностью прожитый день от ещё идущего. Сами события этой границы не показывают: и посреди дня D, и сразу после его конца их позднейший день равен D. Если собрать сессии посреди дня, визит, начатый в 23:50, останется обрезанным: завершённый день слой повторно не собирает. Поэтому последний прожитый день равен первому дню событий плюс позиция минус один (sql/dds/session_scope.sql). Оговорка про «доехал» остаётся в силе и здесь: за хвост потока отвечает не позиция, а «Готовность источника».

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

Первый шаг dds_transform проходит два барьера за общий срок в 300 секунд. Сначала клиент confluent-kafka сверяет подтвержденные смещения группы clickstream_hits с верхними смещениями обоих разделов hits. Группа, оба раздела и все смещения должны существовать. Суммарное отставание должно стать нулевым. Каждый опрос пишет в журнал оба смещения и отставание по разделу.

Только после этого шаг проверяет обе ноды ClickHouse через clusterAllReplicas: у ods.event_dist в system.distribution_queue не должно остаться файлов. Каждый опрос пишет в журнал ноды, число файлов и last_exception очереди. Оба барьера опрашиваются раз в три секунды и делят один крайний срок. После Kafka срок не отсчитывается заново.

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

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

Партиция не заменяется простой вставкой. День собирается в таблицу-двойник — вставка через Distributed раскладывает строки по шардам, — затем REPLACE PARTITION на кластере атомарно подменяет кусок каждого шарда. Эта механика — часть самих трансформаций и строится этапом 4: без неё дневной прогон не повторяем. Сущности, пересчитываемые целиком, идемпотентны и без замены: полная перезапись в ReplacingMergeTree — так живут карта идентичностей и витрина daily_traffic (слой DM).

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

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

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

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

Не входит

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

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

• Задачи трансформаций прямо в работниках пульта. Один зелёный прогон — весь день, но пульт — про ось времени: работников два, шаги дублировались бы, а этапу 5 пришлось бы выдирать их обратно.
• Make-цель или SQL рукой при приёмке. Второй оркестратор рядом с Airflow — против ADR 0013, — а в живом режиме слои копили бы отставание молча.
• Временный даг с нейтральным именем до этапа 5. Дневная сборка вперёд никуда не денется, переезд был бы переименованием ради выдуманной границы.
• Диапазон дней параметрами триггера. Ручному запуску менти параметры передавать некому, а перезапуск после падения требовал бы помнить conf упавшего прогона.
• Принять гонку с хвостом hits. Дыра всплыла бы в DQ-сверке, а чиниться — не раньше этапа 5.

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

• ADR 0003 — зелёный конец графа не переживает отказ выше: отсюда ожидания в триггерах.
• ADR 0009 — пульт мира: его работник живого дня и дёргает конвейер.
• ADR 0013 — жизненный цикл мира: почему прикладным процессом владеет Airflow и почему world_initialize кончается собранными слоями.
• Резолюция — тикет #134 карты #127; что пересобирает каждая сущность — заказ, сессия, карта идентичностей.