diff --git a/dags/dds_transform.py b/dags/dds_transform.py index 116aac4..81911e1 100644 --- a/dags/dds_transform.py +++ b/dags/dds_transform.py @@ -1,8 +1,12 @@ """Трансформации DDS: заказы, сессии и карта идентичностей. -Заказы и сессии собираются в донор, затем публикуются заменой дневных партиций. -Карта идентичностей пересчитывается целиком после заказов и записывается новой -версией; публичное представление выбирает текущие строки через FINAL. +Запросы рассчитывают заказы и сессии за выбранные дни и записывают результат +в промежуточные таблицы. Затем готовые дневные разделы этих таблиц заменяют +разделы основных таблиц. Пока идет расчет, читатели видят прежние данные. + +Карта идентичностей пересчитывается целиком после заказов. Новые строки +добавляются с меткой времени сборки; представление с FINAL выбирает последнюю +версию каждой строки. Даг запускается без расписания и параметров. Даты он выбирает по состоянию таблиц и позиции мира. Новый запуск заново определяет диапазон: наличие даты @@ -252,16 +256,11 @@ def _days(first_day: datetime.date, last_day: datetime.date) -> list[str]: @task def replacements(scope: dict[str, object]) -> list[dict[str, object]]: - """Разложить дни в аргументы размноженной задачи замены. + """Подготовить параметры отдельной задачи replace для каждого дня. - Шаг существует ради ограничения Airflow: размножать задачу можно - только по целому возврату другой задачи — ни по ключу словаря, ни по - карте поверх него. Отсюда отдельный возврат под один список. - - День едет в запрос подстановкой Jinja, а не параметром ClickHouse: - в ALTER … PARTITION сервер параметр не подставляет — замерено, - отвечает синтаксической ошибкой. Значение при этом не внешнее, его - вернул сам ClickHouse запросом объёма. + expand_kwargs получает список словарей и создает по одному экземпляру + задачи на каждый словарь. Дата подставляется в SQL шаблоном Jinja: + параметр ClickHouse в ALTER ... PARTITION здесь не поддерживается. """ return [{"params": {"day": day}} for day in scope["days"]] @@ -271,8 +270,8 @@ def replacements(scope: dict[str, object]) -> list[dict[str, object]]: schedule=None, start_date=START_DATE, is_paused_upon_creation=False, - # Два прогона разом собирали бы один и тот же двойник: второй затёр бы - # чужую сборку между вставкой и заменой партиций. + # Одновременные запуски используют одни промежуточные таблицы. + # Второй мог бы очистить результат первого до замены партиций. max_active_runs=1, template_searchpath=str(SQL_ROOT), tags=["dds"], @@ -301,8 +300,7 @@ def dds_transform(): query = (SQL_ROOT / "dds" / "order_scope.sql").read_text(encoding="utf-8") hook = ClickHouseHook(clickhouse_conn_id=CLICKHOUSE_CONNECTION) row = hook.get_first(query) - # Пустой ответ означает пустой источник: собирать нечего, и - # зеленеть на этом нельзя — прогон встаёт с внятной причиной. + # Без заказов ODS расчет невозможен: задача сообщает об ошибке. if row is None: raise AirflowException( "в ods.order нет ни одного заказа: собирать нечего; " @@ -337,9 +335,9 @@ def dds_transform(): do_xcom_push=False, ) - # Замена идёт по партиции за раз — другой формы у операции нет. День - # становится отдельной задачей: в интерфейсе видно, сколько дней несёт - # прогон, а повтор после отказа стоит одного дня, а не всей пересборки. + # На каждый день создается отдельная задача замены партиции. + # Неудачную замену можно повторить, пока промежуточная таблица + # сохраняет рассчитанные строки этого запуска. replace = SQLExecuteQueryOperator.partial( task_id="replace", conn_id=CLICKHOUSE_CONNECTION, @@ -365,9 +363,8 @@ def dds_transform(): "проверьте, дошёл ли поток до хранилища" ) first_day, last_day = row - # Пропуск, а не отказ: слой полон по последний прожитый день, а - # сегодняшний ещё идёт. Это обычное состояние повторного прогона, - # и в интерфейсе оно видно пропущенной группой. + # Для каждого прожитого дня уже есть строки сессий. + # Новых дней для расчета нет: группа задач будет пропущена. if first_day is None: raise AirflowSkipException( f"сессии собраны по {last_day} включительно, " diff --git a/dags/dm_transform.py b/dags/dm_transform.py index 8e7210c..d657079 100644 --- a/dags/dm_transform.py +++ b/dags/dm_transform.py @@ -1,9 +1,12 @@ """Трансформации слоя DM: сборка стендовых витрин. -Выручка наследует ритм заказов. Сверка соединяет окно заказов с прожитым -хвостом событий. Обе заменяют дневные партиции. Трафик наследует -ретроспективность карты идентичностей и пересчитывает всю историю новой -версией. Объем каждого прогона выводится из состояния слоев. +Даты для расчета выручки берутся из последнего слепка заказов и списка +пропущенных дней. Сверка покупок с заказами учитывает также прожитые дни +событий и дни, где заказ еще не найден. Запросы записывают результаты +в промежуточные таблицы, затем заменяют дневные разделы основных таблиц. + +Трафик пересчитывается за всю историю: новая связь посетителя с пользователем +магазина меняет число известных пользователей и за прошлые дни. """ from __future__ import annotations @@ -62,10 +65,11 @@ def _days(first_day: datetime.date, last_day: datetime.date) -> list[str]: @task def replacements(scope: dict[str, object]) -> list[dict[str, object]]: - """Разложить дни в аргументы размноженной задачи замены. + """Подготовить параметры отдельной задачи replace для каждого дня. - Отдельный словарь нужен каждому экземпляру ``expand_kwargs``; тот же - прием разобран в ``dds_transform.py`` на замене партиций заказов. + expand_kwargs получает список словарей и создает по одному экземпляру + задачи на каждый словарь. Дата подставляется в SQL шаблоном Jinja: + параметр ClickHouse в ALTER ... PARTITION здесь не поддерживается. """ return [{"params": {"day": day}} for day in scope["days"]] @@ -141,7 +145,7 @@ def dm_transform(): @task_group(group_id="purchase_vs_orders") def purchase_vs_orders(): - """Сверка: окно заказов, прожитый хвост и предварительные дни.""" + """Сверка: изменяемые заказы, прожитые покупки и дни ожидания заказа.""" @task def scope() -> dict[str, object]: diff --git a/docs/architecture/dds/session.md b/docs/architecture/dds/session.md index db24f5e..86b20b8 100644 --- a/docs/architecture/dds/session.md +++ b/docs/architecture/dds/session.md @@ -17,10 +17,11 @@ `EventDate` источника, дата по часовому поясу счетчика. У всех событий одной сессии она одинакова. Пересчитывать ее из времени UTC не нужно. -Оба окна нарезки используют порядок `(event_time, event_id)`: при равном -времени события идут по идентификатору. Одинаковый порядок нужен, чтобы -признак начала визита и накопленный номер визита относились к одним -и тем же событиям. +SQL рассчитывает границы визитов двумя оконными функциями: первая отмечает +начало нового визита, вторая по этим отметкам присваивает событиям номер +визита. Обе используют порядок `(event_time, event_id)`: при равном времени +события идут по идентификатору. Иначе два расчета могли бы по-разному +упорядочить события и отнести отметку начала к разным строкам. ## Идентификатор @@ -73,24 +74,28 @@ Три SQL-файла выполняются последовательно в группе `session` дага `dds_transform`: -1. `session_scope.sql` находит первый день событий, которого еще нет - в таблице сессий. Последний день для сборки — первый день событий ODS - плюс `world_position` минус один. Так в расчет попадают только полностью +1. `session_scope.sql` находит первую дату событий, за которую еще нет + сессий. Последний день для расчета — первый день событий ODS + плюс `world_position` минус один; `world_position` — номер следующего + дня для генерации. Так в расчет попадают только полностью прожитые дни. При пустом ODS запрос не возвращает строк и задача падает; если пропущенных прожитых дней нет, `first_day` равен `NULL` и группа пропускается. -2. `session_rebuild.sql` очищает `dds.session_stage_rep` и собирает весь - выбранный диапазон в `dds.session_stage_dist`. Это донор — служебная - таблица для подготовки результата до публикации. Вставка ждет доставки +2. `session_rebuild.sql` очищает промежуточную таблицу + `dds.session_stage_rep`. Затем рассчитывает сессии выбранных дней + и записывает строки через `dds.session_stage_dist`. Пока идет расчет, + в основной таблице остаются прежние данные. Вставка ждет доставки строк на шарды благодаря `distributed_foreground_insert = 1`. -3. `session_replace.sql` заменяет каждый день в `dds.session_rep` партицией - донора. Для каждого дня Airflow создает отдельную задачу `replace`. +3. `session_replace.sql` заменяет в `dds.session_rep` раздел за каждый + выбранный день готовым разделом `dds.session_stage_rep`. Для каждого + дня Airflow создает отдельную задачу `replace`. -В обычном ходе сессии прожитого дня больше не меняются. Если пропущен ранний +Когда все события завершенного дня приняты в ODS, их набор на стенде +больше не меняется. Поэтому готовые сессии обычно не пересчитываются. Если пропущен ранний день, диапазон включает и все последующие прожитые дни, даже уже собранные. -Поиск по наличию даты не обнаружит неполный день, у которого часть строк -сохранилась. Условия повторного запуска и границы атомарности замены -описаны в [справочнике конвейера](../etl.md#идемпотентность). +Если за день сохранилась хотя бы одна сессия, поиск пропущенных дней +не обнаружит потерю остальных. Условия повторного запуска и замены +разделов описаны в [справочнике конвейера](../etl.md#идемпотентность). Сессий еще идущего дня в модели нет; его события доступны через `dds.event_v`. Перед выбором дней даг ждет завершения доставки событий @@ -103,10 +108,10 @@ из `ods.event_v`. Эталон читается из ODS, потому что `dds.event_v` скрывает `VisitID`. -Нулевое число расхождений подтверждает совпадение счета, но не всех -границ и атрибутов сессий. Две ошибки внутри одного посетителя и дня -могут сохранить число визитов. Поэтому в заданиях вы дополнительно -сопоставляете границы сессий по исходным событиям и проверяете страницы, +Нулевое число расхождений подтверждает только число сессий каждого +посетителя за день. Например, ошибочное объединение двух визитов +и разделение другого на два сохранят их общее число. Поэтому в заданиях +вы дополнительно сопоставляете границы сессий по исходным событиям и проверяете страницы, UTM, устройство и город. ## Читатели diff --git a/docs/architecture/dm.md b/docs/architecture/dm.md index b40c376..056b350 100644 --- a/docs/architecture/dm.md +++ b/docs/architecture/dm.md @@ -13,9 +13,11 @@ повторного расчета. Таблицу наполняет `dm_transform`. У такой витрины есть локальная `_rep`, распределенная `_dist` и публичное -представление `_v` с явным списком колонок. Для замены дневных партиций -служат еще две таблицы: `_stage_rep` и `_stage_dist`. Это донор — место -сборки результата перед публикацией. Подробности — в +представление `_v` с явным списком колонок. Для обновления по дням +нужна промежуточная таблица `_stage_rep`, в которую пишут через `_stage_dist`. +В нее записывают рассчитанные строки, пока основная таблица хранит прежние. +Затем дневные разделы основной таблицы заменяют готовыми разделами +промежуточной. Такие разделы называются партициями. Подробности — в [справочнике хранилища](storage.md#имена-объектов). Если запрос только переименовывает или дополняет поля и сохраняет одну @@ -54,23 +56,25 @@ Дни для пересчета выбирают по тому, как меняется источник: - События и сессии прожитого дня на стенде больше не меняются. Витрины - по ним собирают диапазон от первого пропущенного дня до последнего + по ним рассчитывают показатели от первого пропущенного дня до последнего прожитого и заменяют дневные партиции. - Заказы могут меняться в течение семи модельных дней. Последний слепок ODS задает диапазон дат для пересборки; пропущенные дни могут расширить его. -- Новые связи карты идентичностей меняют трактовку прошлых визитов: - сегодняшняя покупка позволяет узнать пользователя и в ранних днях. +- Новая связь в карте идентичностей позволяет определить пользователя + по его прежним визитам: один `client_id` мог встречаться и до покупки, + которая связала его с `user_id`. Поэтому `daily_traffic`, как и карта, пересчитывается целиком. `daily_traffic` каждый раз добавляет всю историю новой версией в `ReplacingMergeTree(_load_ts)`. Дневных партиций у нее нет, а текущие строки выбирает представление через `FINAL`. -У `purchase_vs_orders` два источника. Левая граница — самая ранняя -из трех дат: начало окна последнего слепка, первый пропущенный день -и первый день с `awaiting_order`. Последний нужен потому, что строка -ожидания уже есть в витрине и поиском пропусков не обнаружится. -Правая граница — более поздняя из конца окна заказов и последнего +У `purchase_vs_orders` два источника. Начало диапазона — самая ранняя +из трех дат: первая дата оформления заказа в последнем слепке, первый +пропущенный день и первый день с `awaiting_order` — покупкой, для которой +заказ пока не поступил. День ожидания проверяется отдельно: строки за него +уже есть в витрине, поэтому он не считается пропущенным. +Конец диапазона — более поздняя из последней даты заказов в слепке и последнего прожитого дня с покупкой. Один запуск пересобирает весь отрезок, даже если с предыдущего запуска прошло больше семи модельных дней. @@ -81,22 +85,25 @@ ## Витрины стенда -Публичные представления перечисляют деловые колонки и координаты запуска -`_load_id`, `_load_ts` явно. Новая колонка физической таблицы сама в договор не -попадет. +Представления `_v` явно перечисляют поля для отчетов, а также `_load_id` +и `_load_ts` — идентификатор запуска и время записи. Новая колонка +физической таблицы появится в представлении только после изменения его SQL. ### `dm.revenue_daily_v` -Одна строка — день заказа и категория товара. Витрина читает только оплаченные -заказы из `dds.order_v`, синхронно разворачивает три массива позиций и узнает -категорию артикула через `dic.products`. Один заказ с несколькими позициями -категории считается в `orders` один раз. `units` складывает количество единиц, -`revenue` складывает цену позиции × количество. Скидка и доставка -в выручку не входят; итог заказа между категориями не делится. `aov` равен `revenue / orders` в том же срезе и -хранит два знака после запятой. +Одна строка содержит итоги оплаченных заказов за день по категории товара. +Запрос читает `dds.order_v` и разворачивает массивы артикулов, количеств +и цен в строки позиций заказа. Артикул, количество и цена с одним индексом +становятся одной строкой. Категория артикула берется из `dic.products`. -Первый запуск строит всю доступную историю. Следующие берут границы дней из -последнего слепка ODS: только источник знает, какие даты еще может изменить. +`orders` считает разные заказы: несколько позиций одной категории +в заказе не увеличивают этот счетчик. `units` — сумма количества товаров, +`revenue` — сумма произведений цены на количество. Скидка и доставка +в нее не входят. `aov` равен `revenue / orders` для того же дня и категории +с округлением до двух знаков после запятой. + +Первый запуск рассчитывает выручку за всю доступную историю. Следующие +берут диапазон дат оформления заказов из последнего слепка ODS. Если в DDS есть оплаченные заказы за день, отсутствующий в витрине, левая граница расширяется до него. Агрегат читает DDS; ODS отвечает только на вопрос об объеме. @@ -107,8 +114,9 @@ возвращает `first_day` и `last_day`. Без заказов в ODS или DDS запрос не возвращает строк, и задача завершается ошибкой. 2. [`revenue_daily_rebuild.sql`](../../sql/dm/revenue_daily_rebuild.sql) - очищает донор и одной вставкой собирает весь диапазон в - `dm.revenue_daily_stage_dist`. Вставка ждет доставки на шарды. + очищает `dm.revenue_daily_stage_rep`, рассчитывает показатели за выбранные + дни и записывает строки через `dm.revenue_daily_stage_dist`. + Вставка ждет доставки строк на шарды. 3. [`revenue_daily_replace.sql`](../../sql/dm/revenue_daily_replace.sql) заменяет в `dm.revenue_daily_rep` каждую дневную партицию `report_date` готовой партицией `dm.revenue_daily_stage_rep`. @@ -116,33 +124,38 @@ Повтор замены не добавляет копий строк. Ограничения при частичном отказе описаны в [конвейере](etl.md#идемпотентность). -Цель и донор шардируются -по `cityHash64(product_category)`. Кроме дня, у витрины только категория, а -шесть категорий стартового мира близки по объему. Более сложный составной ключ -распределил бы эту маленькую таблицу без нового урока. +Основная и промежуточная таблицы распределяют строки по +`cityHash64(product_category)`. На каждый день приходится не больше шести +категорий стартового мира. Для такой маленькой таблицы достаточно простого +ключа по категории. ### `dm.purchase_vs_orders_v` -Одна строка — один заказ. Покупки из `dds.event_v` сначала сворачиваются по -`purchase_id[1]`: контракт источника обещает один номер и одну объявленную -сумму. Отдельный счет событий позволяет обнаружить повтор покупки -и присвоить класс `duplicate_event`. Заказы приходят из `dds.order_v`. Обе стороны записаны -подзапросами, затем соединены `FULL OUTER GLOBAL JOIN`: левая -Distributed-таблица без подзапроса вернула бы каждый непарный заказ -по разу с каждого шарда +Одна строка сопоставляет события покупки с заказом магазина. События +из `dds.event_v` группируются по номеру заказа `purchase_id[1]`: +по формату источника у покупки один номер и одна заявленная сумма. +Отдельный счет событий позволяет обнаружить повтор покупки +и присвоить класс `duplicate_event`. Заказы берутся из `dds.order_v`. + +Полное соединение `FULL OUTER GLOBAL JOIN` сохраняет и покупки без заказа, +и заказы без покупки. Обе стороны записаны подзапросами: при прямом чтении +левой Distributed-таблицы каждый непарный заказ повторился бы по числу шардов ([исследование](../research/2026-08-22-distributed-join.md)). -Непарная сторона остается `NULL` благодаря `join_use_nulls = 1`. При -`join_use_nulls = 0` ClickHouse заполнил бы ее нулем или пустой строкой и скрыл -отсутствие источника. Клиентская `declared_revenue` и серверная `items_total` -сравниваются как `Decimal(18, 2)`. Классы проверяются по порядку: +Если для строки нет пары, поля отсутствующей стороны получают `NULL` +благодаря `join_use_nulls = 1`. При `join_use_nulls = 0` они заполнились бы +нулями или пустыми строками, которые сложнее отличить от данных источника. +Клиентская `declared_revenue` и серверная `items_total` сравниваются +как `Decimal(18, 2)`. Классы проверяются по порядку: `awaiting_order` для покупки без заказа, затем `cancelled`, `lost_event`, `duplicate_event`, `amount_delta`, `match`. Поэтому отмененный заказ без события остается отмененным, а не потерянным. -Партиции витрины разделены по `order_day`, цель и донор шардируются по -`cityHash64(order_id)`. Диапазон собирается в доноре одной вставкой и -заменяется по дням. Первичная сборка охватывает всю историю обоих источников; +Партиции витрины разделены по `order_day`. Основная и промежуточная таблицы +распределяют строки по `cityHash64(order_id)`. Запрос рассчитывает весь +выбранный диапазон и одной вставкой записывает строки в промежуточную +таблицу. Затем ее дневные разделы заменяют разделы основной таблицы. +Первичная сборка охватывает всю историю обоих источников; следующие прогоны выбирают дни по правилам обоих источников, описанным выше. ### `dm.daily_traffic_v` @@ -150,8 +163,9 @@ Distributed-таблица без подзапроса вернула бы ка В строке один день. `visitors` считает разные `client_id` из `dds.session_v`; `known_users` считает разные `user_id` после левого соединения с полной текущей `dds.identity_map_v`. Анонимная сессия остается в витрине. -Несколько кук одного человека увеличивают `visitors`, но дают одного -известного пользователя, поэтому `known_users <= visitors`. +Несколько кук, связанных с одним пользователем магазина, учитываются +в `visitors` по отдельности, а в `known_users` дают одного пользователя. +Поэтому `known_users <= visitors`. Сессии и карта распределены по `client_id`. Однако на ClickHouse 26.3 чтение через публичные `_v` с `distributed_product_mode = 'local'` @@ -176,8 +190,8 @@ Distributed-таблица без подзапроса вернула бы ка - `failed_rows` — число отсутствующих или различающихся ключей; - `status` — `pass` при нуле нарушений, иначе `fail`. -Служебные `_load_id` и `_load_ts` называют собравший запуск. Представление -перечисляет колонки явно. Порогов и промежуточного состояния нет. +В `_load_id` и `_load_ts` записаны идентификатор запуска и время расчета +сводки. Даже одно расхождение дает `fail`. ## Звено DQ конвейера @@ -201,29 +215,30 @@ Distributed-таблица без подзапроса вернула бы ка сравниваются с прямыми агрегатами `dds.session_v` и `dds.identity_map_v`. -Область каждой проверки выводится из объединения сторон. Полностью потерянный -день или ключ поэтому не исчезает из результата. Разные ошибки тоже не могут -скомпенсировать друг друга общим итогом: `failed_rows` считается после -сопоставления деловых ключей. +Каждая проверка берет дни и ключи из обоих сравниваемых наборов данных. +Поэтому потерянная строка остается видна на другой стороне сравнения. +Расхождения считаются по ключам: завышение одного показателя не скроет +занижение другого в общей сумме за день. `classes_vs_inventory` сравнивает результат с паспортом мира. Это техническая проверка доставки и обработки эталонных данных стенда. Она не проверяет сам генератор; у настоящего магазина такого эталона нет. Четыре SQL-сверки считают дневные итоги в ClickHouse и возвращают Airflow только -сводку и малый срез диагностики. Техническая сверка с паспортом читает небольшой -файл JSON в Python. Все пять сборок пишут в общий донор. После их успеха даг -заменяет проверенные дневные партиции. Затем именованные задачи читают -опубликованную сводку и падают при `failed_rows > 0`. Расхождение успевает -сохраниться до красного завершения, а ошибка сборки не затирает прежний -результат. Красная задача печатает не больше двадцати нарушенных ключей за -проверку со значениями сторон. Для полного списка итоговую агрегацию SQL-сверки +сводку и несколько примеров расхождений. Техническая сверка с паспортом читает небольшой +файл JSON в Python. Результаты всех пяти проверок записываются в общую +промежуточную таблицу. Когда расчеты завершились, ее дневные разделы +заменяют разделы `dm.dq_summary_rep`. Затем отдельные задачи читают +сохраненную сводку и завершаются ошибкой при `failed_rows > 0`. +Так расхождение сохраняется в таблице до отказа задачи. Если сам расчет +не выполнен, прежняя сводка остается на месте. Задача с расхождениями +выводит в журнал не больше двадцати ключей со значениями обеих сторон. Для полного списка итоговую агрегацию SQL-сверки заменяют на `SELECT * FROM comparison WHERE failed ORDER BY data_date, business_key`. -Если после починки день исчез с обеих сторон, его больше нет и в новом доноре. -Такой день удаляется из сводки отдельно: старая красная строка не должна -превращаться в историю попыток. +Если день исчез из обоих сравниваемых наборов, новый расчет не создаст +для него строк. Даг отдельно удаляет прежний результат за этот день +из сводки. ## Даг самопроверки diff --git a/docs/architecture/etl.md b/docs/architecture/etl.md index 90463ef..cd1fe9d 100644 --- a/docs/architecture/etl.md +++ b/docs/architecture/etl.md @@ -1,21 +1,23 @@ # Дневной конвейер трансформаций Даг `etl_pipeline` собирает модель DDS, обновляет витрины DM и проверяет -результат. Здесь описаны запуск, выбор дней и публикация данных. +результат. Здесь описаны запуск, выбор дней и запись результата в таблицы. Состав моделей — в [справочниках DDS](dds/README.md) и [DM](dm.md). ## Форма `etl_pipeline` последовательно запускает `dds_transform`, `dm_transform` и `dq_check`. Каждый следующий даг начинается после успеха предыдущего. -Для этого `TriggerDagRunOperator` ждет завершения созданного им запуска; -его ошибка завершает ошибкой и соответствующую задачу конвейера. +Для этого `TriggerDagRunOperator` ждет завершения запущенного им дага. +Если тот завершился ошибкой, задача ожидания в конвейере тоже завершится +ошибкой. Поведение проверено через Context7 по документации Airflow 3.x (3.1.6). У этих дагов нет расписания. Их вызывает конвейер; для отдельного разбора можно запустить даг слоя вручную. У каждого разрешен один активный запуск: -два запуска одного дага не должны одновременно заполнять общий донор. -Внутри дага модели собраны в отдельные задачи и группы задач. +два запуска одного дага используют одни промежуточные таблицы и могли бы +стереть результаты друг друга. Расчет каждой модели выполняется отдельной +задачей или группой задач. ## Три пути запуска @@ -40,9 +42,11 @@ world_next_day ─────> источники без ETL Объем прогона — дни, которые обрабатывает один запуск. Даты передавать не нужно: даги определяют их по состоянию таблиц и `world_position`. -Например, после генерации семи дней без ETL один запуск соберет все семь. +Например, после генерации семи дней без ETL один запуск рассчитает сессии +за все семь дней. -Сессии собираются от первого отсутствующего дня до последнего прожитого. +Для расчета сессий выбираются дни от первого, за который в DDS еще нет +сессий, до последнего прожитого. `world_position` хранит номер следующего дня для генерации, поэтому последний прожитый день равен `min(EventDate) + world_position - 1`. Начало календаря берется из ODS; события ранних дней там не удаляются по TTL. @@ -52,9 +56,10 @@ world_next_day ─────> источники без ETL Для заказов диапазон задает последний принятый слепок ODS. Среди строк с максимальной `snapshot_date` выбираются самая ранняя и самая поздняя -даты оформления заказа. Эти дни еще могут измениться. Если в DDS -отсутствует более ранний день, который есть в ODS, левая граница -расширяется до него. Все дни отрезка пересобираются. +даты оформления заказа. Состояние заказов за эти дни еще может измениться. +Если в ODS есть заказы за более ранний день, а в DDS за него нет строк, +начало диапазона сдвигается до этого дня. Запрос заново рассчитывает +заказы за все дни выбранного отрезка. Позиция мира для выбора дней заказов не подходит: с событиями дня D приходит слепок за D−1. Объем рассчитывается по принятым строкам ODS; @@ -62,8 +67,8 @@ world_next_day ─────> источники без ETL Карта идентичностей пересчитывается целиком. Витрины выбирают дни по своим источникам — см. [ритм пересборки DM](dm.md#ритм-пересборки). -Поиск отсутствующих дней проверяет только наличие даты: частично -заполненный день требует отдельной проверки. +Если за день сохранилась хотя бы одна строка, запрос не считает его +пропущенным. Полноту данных проверяют отдельно. ## Готовность источника @@ -96,17 +101,29 @@ world_next_day ─────> источники без ETL не добавляет копии деловых строк. Служебные `_load_id` и `_load_ts` при новой сборке меняются. -Для дневных таблиц порядок такой: `scope` выбирает даты, `rebuild` очищает -донор и собирает в него весь диапазон, затем отдельные задачи `replace` -публикуют дни через `REPLACE PARTITION`. Донор — служебная таблица с той же -структурой и раскладкой по шардам, что у цели. Запись в `_stage_dist` -с `distributed_foreground_insert = 1` дожидается доставки строк перед заменой. +Результат дневного расчета сначала записывается в промежуточную таблицу. +Она нужна, чтобы читатели продолжали видеть прежние данные, пока новые +еще рассчитываются. Например, для сессий это `dds.session_stage_rep`, +а основная таблица — `dds.session_rep`. -Замена выполняется на локальных `_rep` с `ON CLUSTER`. Она атомарна -на каждом шарде: читатель видит старую или новую партицию. Единой -транзакции для двух шардов и всех дней нет. Если замена прошла только -на одном шарде, другой может сохранить прежние данные. Повтор задачи -`replace` с тем же готовым донором повторит замену без добавления копий. +Задачи выполняются в таком порядке: + +1. `scope` выбирает даты для расчета. +2. `rebuild` очищает промежуточную таблицу, рассчитывает строки за выбранные + дни и записывает их через `_stage_dist`. Настройка + `distributed_foreground_insert = 1` заставляет вставку дождаться + доставки строк на шарды. +3. `replace` заменяет в основной таблице раздел за каждый выбранный день + готовым разделом промежуточной таблицы. Раздел таблицы называется + партицией; для замены используется `REPLACE PARTITION`. + +Структура и распределение строк по шардам у двух таблиц должны совпадать. +Замена выполняется на локальных `_rep` с `ON CLUSTER`. На каждом шарде +читатель видит либо старый, либо новый раздел целиком — это атомарность +замены. Общей транзакции для двух шардов и всех дней нет: при сбое один +шард может уже содержать новый результат, а другой — прежний. Повторить +`replace` можно, пока промежуточная таблица хранит тот же результат; +копии строк в основной таблице от этого не появятся. Новый запуск всего конвейера заново выбирает дни. Он не обязан повторить все дни неудачного запуска: если дата уже есть в целевой таблице, поиск diff --git a/docs/architecture/storage.md b/docs/architecture/storage.md index 7d14ed6..51d8a9e 100644 --- a/docs/architecture/storage.md +++ b/docs/architecture/storage.md @@ -95,20 +95,24 @@ DDS: представление читает события ODS. Само совпадение ключей еще не гарантирует локального выполнения запроса через представления — пример есть в [описании `daily_traffic`](dm.md#dmdaily_traffic_v). -Для дневной пересборки DDS и DM используют донор — служебную таблицу, -в которой готовят результат перед заменой партиций. Например, -`dds.order_stage_rep` объявлена через `AS dds.order_rep`, а запись -в нее идет через `dds.order_stage_dist` с тем же `cityHash64(order_id)`. -У цели и донора должны совпадать структура, ключи партиции и сортировки, -первичный ключ и политика хранения. Эти требования сверены через Context7 -с [документацией ClickHouse](https://clickhouse.com/docs/sql-reference/statements/alter/partition#replace-partition) -для используемой схемы на ClickHouse 26.3. Поэтому донор объявлен через -`AS` цели с совпадающими ключами. DDL показан в +Для дневной пересборки DDS и DM результат сначала записывают +в промежуточную таблицу. Основная таблица в это время остается доступна +для чтения с прежними данными. Например, заказы записываются через +`dds.order_stage_dist` в `dds.order_stage_rep`, затем ее дневные разделы +заменяют разделы основной `dds.order_rep`. + +У промежуточной и основной таблиц должны совпадать структура, ключи +партиции и сортировки, первичный ключ и политика хранения. Поэтому +`dds.order_stage_rep` объявлена через `AS dds.order_rep` с совпадающими +ключами. Требования сверены через Context7 с +[документацией ClickHouse](https://clickhouse.com/docs/sql-reference/statements/alter/partition#replace-partition) +для схемы на ClickHouse 26.3. DDL показан в [`sql/ddl/45-dds-tables.sql`](../../sql/ddl/45-dds-tables.sql). -`REPLACE PARTITION` копирует готовую партицию между локальными таблицами -на каждом шарде; заново распределять строки по ключу цели эта операция -не будет. Поэтому донор должен сразу иметь ту же раскладку, что и цель. +Промежуточная таблица должна также распределять строки по тому же ключу, +здесь — `cityHash64(order_id)`. `REPLACE PARTITION` копирует раздел между +локальными таблицами каждого шарда и не перераспределяет строки. +Поэтому уже в промежуточной таблице заказ должен находиться на нужном шарде. Перед заменой вставка с `distributed_foreground_insert = 1` дожидается доставки данных на шарды. diff --git a/sql/dds/session_rebuild.sql b/sql/dds/session_rebuild.sql index 3939858..e4f8ac7 100644 --- a/sql/dds/session_rebuild.sql +++ b/sql/dds/session_rebuild.sql @@ -1,16 +1,19 @@ --- DDS: сборка сессий выбранных дней в донор dds.session_stage_dist. --- Донор очищается целиком; одна вставка собирает весь диапазон. +-- DDS: расчет сессий за дни от first_day до last_day включительно. +-- Сначала очищаем промежуточную таблицу session_stage_rep, затем записываем +-- в нее рассчитанные строки через session_stage_dist. Основная session_rep +-- остается прежней до выполнения session_replace.sql. -- -- Первое окно помечает событие единицей, если с предыдущего события того же -- посетителя за тот же день прошло больше 1800 секунд. Второе окно суммирует --- эти единицы: получается номер визита внутри посетителя и дня. --- PARTITION BY (client_id, event_date) разделяет визиты по модельной полночи. +-- эти единицы: получается номер визита для каждого посетителя за день. +-- PARTITION BY (client_id, event_date) считает каждый модельный день отдельно: +-- события разных дней не попадут в одну сессию даже при короткой паузе. -- Учитываются все типы событий. -- -- У первого события lagInFrame возвращает его собственное время: разница -- равна нулю, opens_visit = 0, и первый visit_no тоже равен нулю. -- Оба окна используют (event_time, event_id), чтобы при равном времени --- признак начала и накопленный номер визита относились к одним событиям. +-- отметка начала и номер визита относились к одним событиям. -- Порог 1800 секунд — правило визита; он не передается параметром. ALTER TABLE dds.session_stage_rep ON CLUSTER clickstream_cluster DROP PARTITION ALL; diff --git a/sql/dds/session_replace.sql b/sql/dds/session_replace.sql index b854ef5..de6bd05 100644 --- a/sql/dds/session_replace.sql +++ b/sql/dds/session_replace.sql @@ -1,7 +1,8 @@ --- DDS: замена одного дня в session_rep готовой партицией session_stage_rep. +-- DDS: заменяем раздел session_rep за выбранный день готовым разделом +-- промежуточной session_stage_rep. Раздел таблицы называется партицией. -- ON CLUSTER выполняет замену на локальных таблицах каждого шарда. --- Замена атомарна на одном шарде, общей транзакции для кластера нет. --- Повтор с тем же донором не добавляет копий строк. +-- На одном шарде раздел меняется целиком; общей транзакции для кластера нет. +-- Повтор с теми же строками session_stage_rep не добавляет копий в session_rep. -- -- Дату из session_scope.sql подставляет Jinja: параметр ClickHouse в -- ALTER ... PARTITION здесь не поддерживается. Поиск дней следующего diff --git a/sql/dds/session_scope.sql b/sql/dds/session_scope.sql index 37f3b7b..10491d6 100644 --- a/sql/dds/session_scope.sql +++ b/sql/dds/session_scope.sql @@ -1,5 +1,5 @@ -- DDS: дни для сборки сессий, от first_day до last_day включительно. --- Пустой ответ означает пустой ODS; NULL в first_day — отсутствие работы. +-- Без событий ODS строк в ответе нет; NULL в first_day — дней для расчета нет. -- -- position — номер следующего дня для генерации. Последний прожитый день -- равен min(EventDate) + position - 1. В отличие от max(EventDate), эта @@ -13,9 +13,10 @@ WITH (SELECT min(EventDate) FROM ods.event_dist) AS first_source_day, first_source_day + ({position:UInt16} - 1) AS last_lived_day, - -- Ищем первый день ODS, которого нет в DDS: максимум собранного дня - -- пропустил бы более ранний пробел. Частично заполненный день так не найти. - -- GLOBAL сравнивает со списком дней всего DDS, а не одного шарда. + -- Ищем первую дату событий ODS, за которую еще нет сессий DDS. + -- Поиск после max(session_date) пропустил бы более раннюю потерянную дату. + -- Если за день есть хотя бы одна сессия, он уже не считается пропущенным. + -- GLOBAL берет даты сессий со всех шардов. -- minOrNull возвращает NULL, если пропусков нет. ( SELECT minOrNull(day) diff --git a/sql/dm/revenue_daily_rebuild.sql b/sql/dm/revenue_daily_rebuild.sql index e762c7c..a1b2ac1 100644 --- a/sql/dm/revenue_daily_rebuild.sql +++ b/sql/dm/revenue_daily_rebuild.sql @@ -1,7 +1,9 @@ --- DM: сборка выручки за выбранный диапазон в донор. --- DROP PARTITION ALL очищает донор на каждом узле. Вставка ниже дает --- одну строку на день заказа и категорию товара: только оплаченные заказы. --- ARRAY JOIN разворачивает согласованные позиции трех массивов вместе. +-- DM: расчет выручки за дни от first_day до last_day включительно. +-- DROP PARTITION ALL очищает промежуточную revenue_daily_stage_rep. +-- Затем INSERT записывает по одной строке итогов на день и категорию товара. +-- Основная revenue_daily_rep остается прежней до revenue_daily_replace.sql. +-- В расчет входят только оплаченные заказы. ARRAY JOIN превращает артикул, +-- количество и цену с одним индексом в массивах в одну строку позиции заказа. -- Причина выбора DROP вместо TRUNCATE — в sql/dds/order_rebuild.sql. ALTER TABLE dm.revenue_daily_stage_rep ON CLUSTER clickstream_cluster DROP PARTITION ALL; diff --git a/sql/dm/revenue_daily_replace.sql b/sql/dm/revenue_daily_replace.sql index 29d98bc..037a02a 100644 --- a/sql/dm/revenue_daily_replace.sql +++ b/sql/dm/revenue_daily_replace.sql @@ -1,8 +1,9 @@ --- DM: замена дня выручки готовой партицией донора. +-- DM: заменяем раздел revenue_daily_rep за выбранный день готовым разделом +-- промежуточной revenue_daily_stage_rep. Раздел таблицы называется партицией. -- Дату из revenue_daily_scope.sql подставляет Jinja: параметр ClickHouse -- в ALTER ... PARTITION здесь не поддерживается. -- ON CLUSTER выполняет замену на локальных таблицах каждого шарда. --- Замена атомарна на одном шарде, общей транзакции для кластера нет. --- Повтор с тем же донором не добавляет копий строк. +-- На одном шарде раздел меняется целиком; общей транзакции для кластера нет. +-- Повтор с теми же строками промежуточной таблицы не добавляет копий в основной. ALTER TABLE dm.revenue_daily_rep ON CLUSTER clickstream_cluster REPLACE PARTITION '{{ params.day }}' FROM dm.revenue_daily_stage_rep; diff --git a/sql/dm/revenue_daily_scope.sql b/sql/dm/revenue_daily_scope.sql index ca6ec8d..dbff094 100644 --- a/sql/dm/revenue_daily_scope.sql +++ b/sql/dm/revenue_daily_scope.sql @@ -1,6 +1,6 @@ -- DM: диапазон пересборки выручки, first_day и last_day включительно. --- Последний слепок ODS задает даты оформления еще изменяемых заказов. --- Агрегат читает DDS; ODS нужен только для границ диапазона. +-- Последний слепок ODS задает даты оформления заказов, которые еще могут +-- измениться. Саму выручку считаем по DDS; ODS нужен только для выбора дат. -- Левая граница расширяется до первого дня с оплаченными заказами DDS, -- которого нет в витрине. Так первый запуск охватывает историю выручки. -- По наличию даты нельзя обнаружить частично заполненный день.