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

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

Даг etl_pipeline собирает модель DDS, обновляет витрины DM и проверяет результат. Здесь описаны запуск, выбор дней и запись результата в таблицы. Состав моделей — в справочниках DDS и DM.

Форма

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

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

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

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.

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

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

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

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

Перед сборкой 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.

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

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

Трансформация собирает слой по правилу, а не сверяет его с источником. Отсюда три вещи, которых она не заметит, и у всех трёх один адрес — контур 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; что пересобирает каждая сущность — заказ, сессия, карта идентичностей.