From 5f317ba8d2681caf2705abc4add91fa904f7be09 Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Sun, 13 Sep 2026 14:33:39 -0400 Subject: [PATCH] =?UTF-8?q?docs(warehouse):=20=D1=83=D0=BF=D1=80=D0=BE?= =?UTF-8?q?=D1=89=D0=B5=D0=BD=D1=8B=20=D0=BE=D0=B1=D1=8A=D1=8F=D1=81=D0=BD?= =?UTF-8?q?=D0=B5=D0=BD=D0=B8=D1=8F=20=D0=B2=D0=B8=D1=82=D1=80=D0=B8=D0=BD?= =?UTF-8?q?,=20=D1=81=D0=B5=D1=81=D1=81=D0=B8=D0=B9=20=D0=B8=20=D0=BF?= =?UTF-8?q?=D0=B5=D1=80=D0=B5=D1=81=D1=87=D0=B5=D1=82=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Зачем: читателю ступеней 1–4 нужно проследить выбор дней, расчет строки и публикацию результата без расшифровки метафор. Что: согласованы справочники DM, сессий, ETL и два раздела хранилища с комментариями шести SQL-примеров. Уточнены ограничения повторного запуска и выбора атрибутов при равном времени событий. Проверка: make lint, git diff --check, локальные Markdown-ссылки и якоря. SQL без комментариев совпадает; AST Python без строк документации совпадает. Исторические разделы и материалы вне границ задачи сохранены. Refs: https://git.dementev.space/ddmitry/clickstream-data-platform/issues/249 --- dags/dds_transform.py | 23 ++-- docs/architecture/dds/session.md | 170 +++++++++++------------ docs/architecture/dm.md | 222 ++++++++++++++++--------------- docs/architecture/etl.md | 172 ++++++++++++------------ docs/architecture/storage.md | 161 ++++++++++------------ sql/dds/session_rebuild.sql | 63 +++------ sql/dds/session_replace.sql | 15 +-- sql/dds/session_scope.sql | 50 +++---- sql/dm/revenue_daily_rebuild.sql | 18 +-- sql/dm/revenue_daily_replace.sql | 12 +- sql/dm/revenue_daily_scope.sql | 24 ++-- 11 files changed, 440 insertions(+), 490 deletions(-) diff --git a/dags/dds_transform.py b/dags/dds_transform.py index 82bb987..116aac4 100644 --- a/dags/dds_transform.py +++ b/dags/dds_transform.py @@ -1,18 +1,13 @@ -"""Трансформации слоя DDS: сборка моделей. +"""Трансформации DDS: заказы, сессии и карта идентичностей. -Заказ и сессия собираются заменой дневных партиций. Карта идентичностей -пересчитывается целиком после заказа, потому что читает его текущую модель. +Заказы и сессии собираются в донор, затем публикуются заменой дневных партиций. +Карта идентичностей пересчитывается целиком после заказов и записывается новой +версией; публичное представление выбирает текущие строки через FINAL. -Расписания у дага нет и параметров тоже: объём каждого прогона он выводит сам -из состояния слоёв и позиции мира, поэтому один и тот же запуск одинаково -верен из триггера дневного конвейера, с кнопки менти и после падения -(docs/architecture/etl.md, «Объём прогона»). - -Заказ и сессия гасят повтор заменой партиции целиком: день собирается в -таблицу-двойник и подменяет собой партицию цели. Карта пишет полный набор пар -новой версией, а точное чтение даёт FINAL. Замена партиции разобрана в -sql/dds/order_replace.sql; точное чтение карты описано в -sql/ddl/50-dds-views.sql. +Даг запускается без расписания и параметров. Даты он выбирает по состоянию +таблиц и позиции мира. Новый запуск заново определяет диапазон: наличие даты +не подтверждает полноту дня и не гарантирует повтор его сборки после сбоя. +Подробности — docs/architecture/etl.md, «Объём прогона» и «Идемпотентность». """ from __future__ import annotations @@ -298,7 +293,7 @@ def dds_transform(): @task_group(group_id="order") def order(): - """Заказ: дышащее окно плюс дни, которых в слое ещё нет.""" + """Заказ: изменяемые дни последнего слепка и пропущенные дни.""" @task def scope() -> dict[str, object]: diff --git a/docs/architecture/dds/session.md b/docs/architecture/dds/session.md index d418ec1..db24f5e 100644 --- a/docs/architecture/dds/session.md +++ b/docs/architecture/dds/session.md @@ -1,111 +1,113 @@ # Модель сессии в DDS -Учебный результат: менти строит расчётную сущность, которой нет в потоке, — -режет события на сессии по правилу визита, чеканит идентификатор и сверяет -свою нарезку с эталоном. Стенд здесь играет роль сессионизатора: в бою её -играет Метрика, которая режет визиты у себя и с октября 2025 отдаёт готовый -`VisitID` прямо в хитах. Поэтому стенд считает сессии сам, а `VisitID` -генератора использует только как эталон самопроверки — иначе сборка сессий -выродилась бы в `GROUP BY` по чужому полю и центральный урок -слоя пропал бы. Без этого абзаца действия стенда выглядят странно: «зачем -считать то, что уже дано колонкой». +Сессия объединяет события одного визита. На стенде вы учитесь определять +ее границы по событиям, вычислять устойчивый идентификатор и сверять +результат с эталоном. Поэтому сборка читает `dds.event_v`, где нет готового +`VisitID`. Этот идентификатор источника остается в ODS для проверки. ## Зерно -Одна строка — один визит ([CONTEXT.md](../../../CONTEXT.md)): подряд идущие -события одной куки без пауз длиннее 30 минут, не пересекающие границу суток. -В нарезку входят все три типа событий — торговые садятся на свои страницы -и держат сессию наравне с просмотрами. События внутри сессии -упорядочиваются по метке времени события (`UTCEventTime` — опорная колонка -приёма). +Зерно описывает, чему соответствует одна строка: здесь это один +[визит](../../../CONTEXT.md#люди-и-события). В него входят подряд идущие +события одного `client_id` без пауз длиннее 30 минут и без перехода через +модельную полночь. Пауза ровно в 30 минут новую сессию не открывает. +Учитываются все три типа событий: `pageview`, `add_to_cart`, `purchase`. -Граница суток проходит по модельному дню счётчика, и он у события уже есть: -`EventDate` — свойство содержимого. Раз сессия сутки не пересекает, у всех -её событий `EventDate` один; он и становится **днём сессии** — колонкой -партиции, без пересчёта. +День сессии `session_date` берется из `event_date` событий DDS — это +`EventDate` источника, дата по часовому поясу счетчика. У всех событий +одной сессии она одинакова. Пересчитывать ее из времени UTC не нужно. + +Оба окна нарезки используют порядок `(event_time, event_id)`: при равном +времени события идут по идентификатору. Одинаковый порядок нужен, чтобы +признак начала визита и накопленный номер визита относились к одним +и тем же событиям. ## Идентификатор -Идентификатор сессии в природе не существует: сессия — расчётная сущность, -и id ей чеканит тот, кто сессионизирует. `VisitID` — такой же суррогат, -отчеканенный генератором в роли Метрики; наш — `session_id`, -детерминированный хеш от пары «кука + начало сессии» (`cityHash64`). -Детерминированность обязательна: партиция дня пересобирается заменой, и -повторный прогон обязан отчеканить те же id — счётчик или UUID это правило -ломают. +`session_id` вычисляется как `cityHash64(client_id, min(event_time))`. +При повторной сборке тех же событий посетитель и начало визита остаются +прежними, поэтому идентификатор не меняется. Счетчик или случайный UUID +выдали бы другие значения. -`session_id` — не ключ сортировки и не ключ шардирования, а колонка-ручка: -в соединениях он не участвует — события его не несут, мост к заказам живёт -на событиях. Читатели ручки — лаба сессий (сопоставить свою нарезку со -стендовой и назвать расхождение одним числом) и разбор DQ-расхождений. +По `session_id` можно сопоставить сессии до и после пересборки и назвать +конкретную строку при разборе расхождения. Он не используется для сортировки +или шардирования. У событий этого поля нет; связь с заказами строится +по событиям покупки. ## Состав строки -Боевой образец — таблица визитов Метрики, парная к нашему формату хита; по -ней состав и собран: +- `client_id`, `session_id`, `session_date` — посетитель, сессия и ее день. +- `started_at`, `finished_at`, `duration_seconds` — время первого и последнего + события и разница между ними в секундах. Времена хранятся в часовом поясе + счетчика. +- `events_total`, `pageviews`, `cart_adds`, `purchases` — число всех событий + и счетчики трех типов. +- `entry_url`, `exit_url` — страницы первого и последнего события. +- `utm_*` — метки первого события: по ним визит относят к рекламной кампании. +- `device_category`, `region_city` — устройство и город посетителя. + На стенде они постоянны для одной куки. +- `_load_id`, `_load_ts` — идентификатор запуска и время сборки строки. -- **ключи и времена**: кука, `session_id`, начало и конец сессии, - длительность, день сессии. Времена — в поясе счётчика по правилу слоя; -- **счётчики**: событий всего, просмотров, корзин, покупок; -- **страницы**: входная и выходная — как `StartURL`/`EndURL` визита; полного - списка страниц нет и у Метрики, топы считаются по событиям; -- **атрибуция**: UTM-метки первого события сессии — так визит атрибутирует - Метрика. В v1 витрина брала последнее касание — заменено осознанно; -- **паспорт куки**: категория устройства (расшифрованная, язык бизнес-модели) - и город — у куки они постоянны, берутся с первого события; -- **служебные**: `_load_id` и `_load_ts` собравшего запуска — правило слоя - ([модель заказа](order.md), «Служебные колонки»). +В готовом SQL страницы выбираются через `argMin` и `argMax`, а UTM, +устройство и город — через `argMin`, только по `event_time`. При одинаковом +времени выбор между событиями не определен. Полный порядок окон нарезки +сам по себе эту неоднозначность не устраняет. Это ограничение учитывается +в [задании ступени 4](../../labs/4-1-session-pipeline.md). -Осознанные расхождения с образцом: **денег в сессии нет** — визит Метрики -несёт клиентскую выручку, у нас деньги считает только бэкенд (спека, раздел -7), а клиентская сумма живёт в сверке; **целей нет** — их место заняли -счётчики торговых событий. +В сессии нет суммы выручки: ее считают по заказам магазина. Клиентская +сумма покупки используется в сверке с заказами и в учебной UTM-витрине. ## Движок и наполнение -`dds.session_rep` — `ReplicatedMergeTree` без колонки версии, по образцу -заказа: партиция — день сессии, ключ сортировки — кука и начало сессии. -`dds.session_dist` шардирует по `cityHash64(client_id)`, тем же выражением, -что события, — это ко-локация с `dds.identity_map_rep` ([дока -хранилища](../storage.md), «Раскладка по шардам»). -Совпадение ключа с событиями не случайно: все события куки лежат на одном -шарде, поэтому сессионизация локальна по природе — капкан смены ключа при -вставке, который дока хранилища обещает сборкам DDS, у сессий не стреляет. +`dds.session_rep` хранит готовые сессии в `ReplicatedMergeTree` без колонки +версии. При чтении не нужно заново выполнять оконный расчет. Партиция — +раздел таблицы за один `session_date`; ключ сортировки — посетитель +и начало сессии. Читайте результат через `dds.session_v`: представление +над `dds.session_dist` явно перечисляет деловые и служебные колонки. -Сборка — дневной батч: хранилище выше ODS по смыслу батчевое, раз второй -источник приходит раз в сутки. Собирается день, который мир уже прожил, и -собирается один раз и целиком; какой день прожит, слой спрашивает у позиции -мира ([дневной конвейер](../etl.md), «Объём прогона»). Идемпотентность держит -замена партиции. В отличие от заказов, где батч пересобирает до K «дышащих» -дней, у сессии дышащего окна нет: прожитый день самодостаточен (правило -спеки, раздел 6). Второй раз он соберётся только -вынужденно — когда упавшая замена оставила позади дыру и хвост от неё накрыл -соседние дни. Дней в одном прогоне при этом бывает несколько — столько, -сколько мир успел прожить без слоя. Внутри живого дня -модель сессии про сегодня честно молчит — «сегодня» смотрят через -`dds.event_v`; сессия и по смыслу не финальна, пока день не прожит. +Источник сборки — `dds.event_v`. События, сессии и карта идентичностей +распределены по одному ключу `cityHash64(client_id)`: данные посетителя +находятся на одном шарде. Роли объектов и правило записи через `_dist` +описаны в [справочнике хранилища](../storage.md#раскладка-по-шардам). -Публичная поверхность — `dds.session_v`: обычное представление с явным -списком деловых и собственных служебных колонок над `dds.session_dist`. -Роли объектов и доступ следуют общей [конвенции](../storage.md), «Имена -объектов». +Три SQL-файла выполняются последовательно в группе `session` дага +`dds_transform`: -Источник сборки — `dds.event_v` ([контракт события](event.md)): сессии -нужны от него ключи нарезки, тип события, страница, UTM и паспорт куки. -Механика запуска — где собирается партиция и как подменяется на кластере — -не входит: это устройство [дневного конвейера](../etl.md). +1. `session_scope.sql` находит первый день событий, которого еще нет + в таблице сессий. Последний день для сборки — первый день событий ODS + плюс `world_position` минус один. Так в расчет попадают только полностью + прожитые дни. При пустом ODS запрос не возвращает строк и задача падает; + если пропущенных прожитых дней нет, `first_day` равен `NULL` и группа + пропускается. +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`. + +В обычном ходе сессии прожитого дня больше не меняются. Если пропущен ранний +день, диапазон включает и все последующие прожитые дни, даже уже собранные. +Поиск по наличию даты не обнаружит неполный день, у которого часть строк +сохранилась. Условия повторного запуска и границы атомарности замены +описаны в [справочнике конвейера](../etl.md#идемпотентность). + +Сессий еще идущего дня в модели нет; его события доступны через +`dds.event_v`. Перед выбором дней даг ждет завершения доставки событий +в ODS — см. [готовность источника](../etl.md#готовность-источника). ## Сверка с эталоном -Счётная DQ-проверка с зерном «кука × день»: число сессий нашей нарезки -равно `uniqExact(VisitID)` по тем же событиям. Счёт по Distributed ловит и -склеенную, и разрезанную сессию — мимо него проходит лишь взаимная -компенсация ошибок внутри одной куки за день; эталон в модель не тащим — -строка сессии остаётся нашей, а не генераторной. Поштучное сопоставление -границ — материал лабы, где менти разбирает найденное расхождение руками. Дом проверки — -контур DQ этапа 4. Сверке нужен `VisitID`, а сборке сессий — нет; контракт -`dds.event_v` его скрывает, и за эталоном проверка идёт в `ods.event_v` -([контракт события](event.md)). +Проверка `sessions_vs_reference` в `dq_check` сравнивает для каждого +посетителя и дня число строк `dds.session_v` с `uniqExact(VisitID)` +из `ods.event_v`. Эталон читается из ODS, потому что `dds.event_v` +скрывает `VisitID`. + +Нулевое число расхождений подтверждает совпадение счета, но не всех +границ и атрибутов сессий. Две ошибки внутри одного посетителя и дня +могут сохранить число визитов. Поэтому в заданиях вы дополнительно +сопоставляете границы сессий по исходным событиям и проверяете страницы, +UTM, устройство и город. ## Читатели diff --git a/docs/architecture/dm.md b/docs/architecture/dm.md index 4dc98f7..b40c376 100644 --- a/docs/architecture/dm.md +++ b/docs/architecture/dm.md @@ -1,87 +1,83 @@ # Слой DM: витрины -Как модель DDS превращается в витрины для потребителя BI: какую форму -принимает витрина, кто какие витрины строит и как слой проверяется. Состав -колонок каждой витрины задаёт раздел 7 мастер-спеки; здесь — форма, -наполнение и контроль. +Витрины готовят данные DDS для отчетов и дашбордов. Здесь описаны форма +хранения, правила обновления и проверки. Состав колонок задан в +[спецификации](../specs/2026-07-30-stand-v2-realism.md#витрины-dm) +и [учебных заданиях](../labs/README.md). ## Таблица или представление -Дневной агрегат и сверка по заказам меняют зерно источника. На этом стенде такие -результаты хранятся в `dm.<имя>_rep` / `_dist` и читаются через публичное -представление `dm.<имя>_v` с явным списком колонок. Имена и устройство объектов -следуют общим [конвенциям хранилища](storage.md). Таблицы наполняет -`dm_transform`; ритм и механика пересборки описаны ниже. В задании этот путь -приходится пройти целиком: рассчитать агрегат, загрузить его и включить -обновление в конвейер. +Зерно — то, чему соответствует одна строка. Например, в заказах это заказ, +а в дневной выручке — день и категория товара. На стенде результат +с новым зерном хранится в таблице: отчет читает готовые итоги без +повторного расчета. Таблицу наполняет `dm_transform`. -Переименование колонок и обогащение строки зерно не меняют. Такие витрины на -стенде остаются обычными представлениями и не хранят вторую копию событий. +У такой витрины есть локальная `_rep`, распределенная `_dist` и публичное +представление `_v` с явным списком колонок. Для замены дневных партиций +служат еще две таблицы: `_stage_rep` и `_stage_dist`. Это донор — место +сборки результата перед публикацией. Подробности — в +[справочнике хранилища](storage.md#имена-объектов). -На объёмах стенда все витрины могли бы остаться представлениями: условия -проходят сквозь их цепочку +Если запрос только переименовывает или дополняет поля и сохраняет одну +строку на событие, достаточно обычного представления. Оно выполняет +запрос при чтении, не хранит вторую копию событий и не требует пересборки. + +На небольших данных стенда и агрегаты могли бы вычисляться при чтении ([исследование](../research/2026-08-22-view-predicate-pushdown.md)). -Мы все равно оставили материализованные агрегаты ради практики: без них в -программе не было бы загрузки и пересборки витрин. Представления дают другой -случай: преобразование строк на прежнем зерне. +Готовые таблицы нужны в программе для практики загрузки, обновления +и включения витрин в конвейер. ## Что готово в стенде, а что остаётся заданием -Стенд привозит готовыми витрины трёх сюжетов дашборда этапа 8 и -инфраструктуру контроля: +Готовы три витрины дашборда и сводка качества: -- `revenue_daily` — выручка по дням и категориям; образец материализованной - витрины, на него ссылаются учебные задания; -- `purchase_vs_orders` — сверка A+C с классами расхождений; зерно — заказ, - шардирование сохраняет ко-локацию заказов по `cityHash64(order_id)`; -- `daily_traffic` — пара «посетители» / «известные пользователи», обогащение - [картой идентичностей](dds/identity.md); -- `dq_summary` — сводка проверок качества (ниже). +- `revenue_daily` — одна строка на день заказа и категорию товара; + образец дневной пересборки для учебных заданий. +- `purchase_vs_orders` — одна строка на заказ: сопоставление покупки + из кликстрима с заказом магазина и класс расхождения. +- `daily_traffic` — одна строка на день: число посетителей и известных + пользователей с учетом [карты идентичностей](dds/identity.md). +- `dq_summary` — одна строка на день данных и проверку качества. -Остальные витрины спеки нужно построить в учебных заданиях по готовому ТЗ и -образцу. Разминка — представление `events_enriched_v` (зерно события), затем -агрегаты `top_pages_daily`, `session_overview` и `dq_errors_daily`, последней — -самая содержательная `utm_effectiveness` с уроком `declared_*`. В финале -каждого задания с агрегатом его задачу нужно добавить в `dm_transform`. До -этого конвейер о витрине не знает, после — пересобирает её вместе со -стендовыми. Представление из разминки добавлять некуда: его не нужно -пересобирать. Тексты заданий лежат в [`docs/labs/`](../labs/README.md). +В [заданиях ступени 1](../labs/README.md) вы строите `events_enriched_v`, +`top_pages_daily`, `session_overview`, `dq_errors_daily` и `utm_effectiveness`. +Для каждой физической витрины нужно добавить сборку в `dm_transform`. +Представление `events_enriched_v` отдельной задачи не требует. -Дашборд этапа 8 читает только готовые витрины стенда, поэтому его приёмка не -зависит от домашних заданий. Отдельный дашборд по учебным витринам можно -собрать в [факультативном задании](../labs/optional/superset-dashboard.md), -доступном после ступени 2. Устройство стендового дашборда описано в -[справочнике Superset](superset.md). +Стендовый дашборд читает готовые витрины и работает до выполнения заданий. +Дашборд по своим витринам можно собрать в +[факультативе](../labs/optional/superset-dashboard.md) после ступени 2. +Устройство стендового дашборда — в [справочнике Superset](superset.md). ## Ритм пересборки -День витрины пересобирается, пока может измениться его источник, — -поэтому ритмов три, по числу источников: +Дни для пересчета выбирают по тому, как меняется источник: -- **от событий и сессий** — день замерзает сразу; пересборка догоняет - хвост несобранных дней заменой дневных партиций; -- **от заказов** — день дышит, пока входит в последний слепок; пересборка - заменяет партиции этого отрезка, дни за ним закрыты по построению; -- **от карты идентичностей** — не замерзает никогда: склейка - ретроспективна, покупка сегодня делает куку известной во всех её прошлых - днях, а карта пересчитывается целиком. +- События и сессии прожитого дня на стенде больше не меняются. Витрины + по ним собирают диапазон от первого пропущенного дня до последнего + прожитого и заменяют дневные партиции. +- Заказы могут меняться в течение семи модельных дней. Последний слепок ODS + задает диапазон дат для пересборки; пропущенные дни могут расширить его. +- Новые связи карты идентичностей меняют трактовку прошлых визитов: + сегодняшняя покупка позволяет узнать пользователя и в ранних днях. + Поэтому `daily_traffic`, как и карта, пересчитывается целиком. -Поэтому `daily_traffic` наследует у своего источника и ритм, и механику: -полный пересчёт всех дней каждый прогон в `ReplacingMergeTree(_load_ts)`, -как у самой карты. Дневных партиций у неё нет: партиция — единица замены, -а эта витрина заменяется только целиком. На стенде такая работа остается -быстрой; витрина читает сессии и карту через их публичные договоры. -Пересчитывать точечно только затронутые дни не из чего: для этого надо -знать, какие связки новые, а полный пересчёт карты этот след затирает — -восстановление следа оставлено лабе повышенной сложности (issue #149). +`daily_traffic` каждый раз добавляет всю историю новой версией +в `ReplacingMergeTree(_load_ts)`. Дневных партиций у нее нет, а текущие +строки выбирает представление через `FINAL`. -У `purchase_vs_orders` два источника и одна непрерывная пересборка. Заказная -сторона задает изменяемое окно последним принятым слепком ODS, событийная — -последний прожитый день с `purchase`. Левая граница возвращает первую дыру и -самый ранний день с `awaiting_order`: предварительная строка уже есть в -витрине, поэтому поиск отсутствующего дня ее не заметит. Правая граница -доходит до более позднего из двух хвостов. Так один прогон догоняет пакетный -разгон длиннее K и не оставляет старое ожидание замороженным. +У `purchase_vs_orders` два источника. Левая граница — самая ранняя +из трех дат: начало окна последнего слепка, первый пропущенный день +и первый день с `awaiting_order`. Последний нужен потому, что строка +ожидания уже есть в витрине и поиском пропусков не обнаружится. +Правая граница — более поздняя из конца окна заказов и последнего +прожитого дня с покупкой. Один запуск пересобирает весь отрезок, +даже если с предыдущего запуска прошло больше семи модельных дней. + +Для учебной сводки брака действует отдельное правило: дни выбираются +по времени загрузки, а прежние итоги сохраняются после удаления +источника по TTL. Оно разобрано в +[задании `dq_errors_daily`](../labs/1-4-dq-errors-daily.md). ## Витрины стенда @@ -91,34 +87,49 @@ ### `dm.revenue_daily_v` -Зерно: день заказа × категория товара. Витрина читает только оплаченные +Одна строка — день заказа и категория товара. Витрина читает только оплаченные заказы из `dds.order_v`, синхронно разворачивает три массива позиций и узнает категорию артикула через `dic.products`. Один заказ с несколькими позициями категории считается в `orders` один раз. `units` складывает количество единиц, -`revenue` складывает цену позиции × количество. Скидка, доставка и итог заказа -между категориями не делятся. `aov` равен `revenue / orders` в том же срезе и +`revenue` складывает цену позиции × количество. Скидка и доставка +в выручку не входят; итог заказа между категориями не делится. `aov` равен `revenue / orders` в том же срезе и хранит два знака после запятой. Первый запуск строит всю доступную историю. Следующие берут границы дней из последнего слепка ODS: только источник знает, какие даты еще может изменить. -Пропущенный оплаченный день расширяет левую границу, чтобы частичный отказ -не оставил дыру после следующего прогона. Агрегат при этом читает DDS; ODS -отвечает только на вопрос об объеме. +Если в DDS есть оплаченные заказы за день, отсутствующий в витрине, +левая граница расширяется до него. Агрегат читает DDS; ODS отвечает +только на вопрос об объеме. -`dm.revenue_daily_rep` нарезана по `report_date`; день собирается в -таблице-доноре и заменяется через `REPLACE PARTITION`. Цель и донор шардируются +В группе `revenue_daily` дага `dm_transform` три шага: + +1. [`revenue_daily_scope.sql`](../../sql/dm/revenue_daily_scope.sql) + возвращает `first_day` и `last_day`. Без заказов в ODS или DDS + запрос не возвращает строк, и задача завершается ошибкой. +2. [`revenue_daily_rebuild.sql`](../../sql/dm/revenue_daily_rebuild.sql) + очищает донор и одной вставкой собирает весь диапазон в + `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`. + +Повтор замены не добавляет копий строк. Ограничения при частичном отказе +описаны в [конвейере](etl.md#идемпотентность). + +Цель и донор шардируются по `cityHash64(product_category)`. Кроме дня, у витрины только категория, а шесть категорий стартового мира близки по объему. Более сложный составной ключ распределил бы эту маленькую таблицу без нового урока. ### `dm.purchase_vs_orders_v` -Зерно — заказ. Покупки из `dds.event_v` сначала сворачиваются по +Одна строка — один заказ. Покупки из `dds.event_v` сначала сворачиваются по `purchase_id[1]`: контракт источника обещает один номер и одну объявленную -сумму, а отдельный счет событий оставляет класс `duplicate_event` -наблюдаемым. Заказы приходят из `dds.order_v`. Обе стороны записаны -подзапросами, затем соединены `FULL OUTER GLOBAL JOIN`: голая левая -Distributed-таблица вернула бы каждый непарный заказ по разу с каждого шарда +сумму. Отдельный счет событий позволяет обнаружить повтор покупки +и присвоить класс `duplicate_event`. Заказы приходят из `dds.order_v`. Обе стороны записаны +подзапросами, затем соединены `FULL OUTER GLOBAL JOIN`: левая +Distributed-таблица без подзапроса вернула бы каждый непарный заказ +по разу с каждого шарда ([исследование](../research/2026-08-22-distributed-join.md)). Непарная сторона остается `NULL` благодаря `join_use_nulls = 1`. При @@ -129,10 +140,10 @@ Distributed-таблица вернула бы каждый непарный з `duplicate_event`, `amount_delta`, `match`. Поэтому отмененный заказ без события остается отмененным, а не потерянным. -Витрина нарезана по `order_day`, цель и донор шардируются по +Партиции витрины разделены по `order_day`, цель и донор шардируются по `cityHash64(order_id)`. Диапазон собирается в доноре одной вставкой и заменяется по дням. Первичная сборка охватывает всю историю обоих источников; -следующие прогоны идут по двустороннему ритму, описанному выше. +следующие прогоны выбирают дни по правилам обоих источников, описанным выше. ### `dm.daily_traffic_v` @@ -140,28 +151,26 @@ Distributed-таблица вернула бы каждый непарный з `dds.session_v`; `known_users` считает разные `user_id` после левого соединения с полной текущей `dds.identity_map_v`. Анонимная сессия остается в витрине. Несколько кук одного человека увеличивают `visitors`, но дают одного -`known_user`, поэтому `known_users <= visitors`. +известного пользователя, поэтому `known_users <= visitors`. -Сессии и карта лежат по `client_id`, но публичные `_v` скрывают -`Distributed`-таблицы от локальной подстановки. На ClickHouse 26.3 -`EXPLAIN PLAN` показал чтение обоих шардов на инициаторе: настройка -`distributed_product_mode = 'local'` локального соединения не дала. Поэтому -витрина использует явный `GLOBAL LEFT JOIN` через публичные договоры, а не -подменяет их физическими таблицами. Это дороже по сети, зато граница слоев -остается явной. Новая связь куки меняет прошлые дни, поэтому каждый запуск -пишет полную историю новой версией в `ReplacingMergeTree(_load_ts)`, а -публичное представление выбирает текущие строки через `FINAL`. +Сессии и карта распределены по `client_id`. Однако на ClickHouse 26.3 +чтение через публичные `_v` с `distributed_product_mode = 'local'` +не дало локального соединения. Поэтому в запросе явно указан +`GLOBAL LEFT JOIN`. Он сохраняет сессии без известного пользователя; +в `known_users` учитываются только ненулевые `user_id`. +Обновление всей истории новой версией описано в разделе +[«Ритм пересборки»](#ритм-пересборки). ## `dm.dq_summary` -Зерно публичного `dm.dq_summary_v` — «день данных × атомарная проверка». +Одна строка `dm.dq_summary_v` — день данных и одна проверка. Строка отвечает, каково качество дня сейчас; историю запусков хранит Airflow. Партиция совпадает с днём данных и заменяется через `REPLACE PARTITION`. Деловые колонки сводки: - `data_date` — проверяемый день; -- `check_name` — устойчивое имя инварианта; +- `check_name` — имя проверяемого правила; - `reference_rows` — число деловых ключей на эталонной стороне; - `actual_rows` — число деловых ключей в проверяемом объекте; - `failed_rows` — число отсутствующих или различающихся ключей; @@ -173,7 +182,7 @@ Distributed-таблица вернула бы каждый непарный з ## Звено DQ конвейера Даг `dq_check` стоит в `etl_pipeline` после `dm_transform` и проверяет пять -атомарных договоров: +правил: - `sessions_vs_reference`: ключ — `client_id` внутри дня, эталон — `uniqExact(VisitID)` из `ods.event_v`, объект — число строк @@ -197,9 +206,9 @@ Distributed-таблица вернула бы каждый непарный з скомпенсировать друг друга общим итогом: `failed_rows` считается после сопоставления деловых ключей. -`classes_vs_inventory` — техническая растяжка пути от генератора до DM. Она -подтверждает, что до витрины доехал записанный в паспорте мир, но не проверяет -сам генератор: у настоящего магазина такого оракула нет. +`classes_vs_inventory` сравнивает результат с паспортом мира. Это техническая +проверка доставки и обработки эталонных данных стенда. Она не проверяет +сам генератор; у настоящего магазина такого эталона нет. Четыре SQL-сверки считают дневные итоги в ClickHouse и возвращают Airflow только сводку и малый срез диагностики. Техническая сверка с паспортом читает небольшой @@ -218,17 +227,20 @@ business_key`. ## Даг самопроверки -Домашние задания проверяет `homework_check` — отдельный даг без расписания, -который запускают вручную. В нём есть задача для каждого из -[пяти заданий по витринам](../labs/README.md). -Задача красная, пока витрины нет или счёт не сходится, и зелёная, когда -построено верно; сообщение об ошибке говорит, что не так и куда смотреть. Проверки — инварианты против источников витрины, -золотые константы не прибиваются. В `dq_summary` даг не пишет: он инструмент -лабы, а не история качества конвейера. +`homework_check` запускается вручную и содержит задачу для каждого из +[пяти заданий по витринам](../labs/README.md). Если витрины нет или значения +расходятся с источником, задача завершается ошибкой. Журнал показывает +причину и примеры несовпадений. Успех означает, что проверяемые значения +совпали; остальные требования задания вы проверяете сами. -Приём проверен на соседнем учебном стенде Airflow + Greenplum: -валидационный даг с группами проверок по слоям давал обратную связь по каждому -шагу задания, не трогая рабочий конвейер. +Ожидаемые значения рассчитываются по исходным данным текущего стенда. +Для четырех заданий это события или сессии DDS, для сводки брака — +таблицы ошибок ODS. Сравнение по деловому ключу учитывает отсутствующие +строки с обеих сторон. Результаты не пишутся в `dq_summary`: +самопроверка учебных витрин отделена от качества готового конвейера. + +Готовые SQL-проверки открывайте после собственного решения: они содержат +расчет существенной части задания. ## Отклонённые варианты diff --git a/docs/architecture/etl.md b/docs/architecture/etl.md index 0a86a47..90463ef 100644 --- a/docs/architecture/etl.md +++ b/docs/architecture/etl.md @@ -1,28 +1,21 @@ # Дневной конвейер трансформаций -Как принятые источники превращаются в модель DDS и витрины DM: кто запускает -трансформации, когда и в каком объёме. Правила самих сущностей — зерно, -движки, что именно пересобирается — живут в доках слоя -([карта модели DDS](dds/README.md)); здесь — только конвейер. - -Дневная сборка вперёд появляется на этапе 4 и остаётся основным режимом -конвейера. На этапе 5 к ней добавляется партиционная переобработка: запуск -произвольного прошлого дня по команде. +Даг `etl_pipeline` собирает модель DDS, обновляет витрины DM и проверяет +результат. Здесь описаны запуск, выбор дней и публикация данных. +Состав моделей — в [справочниках DDS](dds/README.md) и [DM](dm.md). ## Форма -Дневной конвейер на стенде один — даг `etl_pipeline`. Своей работы -у него нет: он дёргает слойные даги последовательно и ждёт каждый — -`dds_transform`, затем `dm_transform` и `dq_check`. Состав проверок описан в -[доке слоя DM](dm.md). +`etl_pipeline` последовательно запускает `dds_transform`, `dm_transform` +и `dq_check`. Каждый следующий даг начинается после успеха предыдущего. +Для этого `TriggerDagRunOperator` ждет завершения созданного им запуска; +его ошибка завершает ошибкой и соответствующую задачу конвейера. +Поведение проверено через Context7 по документации Airflow 3.x (3.1.6). -Ждущие вызовы сделаны `TriggerDagRunOperator`. Проверка через Context7 по -ближайшей доступной документации Airflow 3.x (версия 3.1.6) подтвердила, что -оператор опрашивает точный созданный им прогон и краснеет вслед за потомком. - -Слойные даги устроены по образцу `orders_ingest`: без расписания, один -активный прогон, запускаются триггером конвейера или рукой при разборе -происшествий. Витрины внутри `dm_transform` — отдельные задачи. +У этих дагов нет расписания. Их вызывает конвейер; для отдельного разбора +можно запустить даг слоя вручную. У каждого разрешен один активный запуск: +два запуска одного дага не должны одновременно заполнять общий донор. +Внутри дага модели собраны в отдельные задачи и группы задач. ## Три пути запуска @@ -34,83 +27,98 @@ world_live_day ───┼─> etl_pipeline ─> dds_transform ─> dm_transfor world_next_day ─────> источники без ETL ``` -- **Живой день.** `world_live_day` дёргает конвейер с ожиданием — последним - шагом, после установки позиции. Порядок несёт смысл: день сыгран — факт, - упавшая трансформация его не отменяет, поэтому позиция ставится раньше; но - зеленеть, не узнав, разложился ли день, шаговый даг не должен ([ADR - 0003](../adr/0003-dag-run-state.md)). Такой отказ чинится перезапуском - `etl_pipeline` рукой — перезапуску ничего не нужно помнить, см. «Объём - прогона». -- **Разгон пачкой.** `world_next_day` конвейер не дёргает: даги модельного - времени управляют генератором и только им. После разгона менти запускает - конвейер сам — и видит, как слои догоняют уехавший вперёд мир. -- **Рождение мира.** `world_initialize` дёргает конвейер последним шагом: - `make up` отдаёт стенд с собранными слоями, а не пустые витрины. Уроки - ручного запуска начинаются со следующих дней. +- `world_live_day` запускает конвейер после успешной генерации, приема + заказов и обновления `world_position`. Если трансформация упала, день + уже прожит и позиция не откатывается. После исправления причины + запустите `etl_pipeline` вручную. +- `world_next_day` отправляет дни пачкой и ждет приема заказов, но конвейер + не запускает. После него запустите `etl_pipeline` сами. +- `world_initialize` запускает конвейер последним шагом: при успешном + первом `make up` начальные данные уже собраны в DDS и DM. ## Объём прогона -Параметров у конвейера нет: объём он выводит сам из `world_position` и -состояния слоёв. Сессии догоняют хвост — от первого несобранного дня по -последний прожитый; заказы пересобирают дышащее окно и тот же хвост; карта -идентичностей пересчитывается целиком; правила витрин — у -[слоя DM](dm.md). -Поэтому один и тот же запуск одинаково верен из триггера, с кнопки менти и -после падения, а разгону на семь дней хватает одного прогона. +Объем прогона — дни, которые обрабатывает один запуск. Даты передавать +не нужно: даги определяют их по состоянию таблиц и `world_position`. +Например, после генерации семи дней без ETL один запуск соберет все семь. -Окно заказов задаёт последний слепок, наблюдаемый в ODS. У каждого заказа -`snapshot_date` равен последнему слепку, в котором источник его показал. -Трансформация выбирает строки с максимальной `snapshot_date` и берёт среди них -минимальный и максимальный `order_date`. Это границы дней, состояние которых -источник ещё может изменить -([ADR 0014](../adr/0014-order-version-snapshot-date.md)). Кроме живого окна, -трансформация закрывает пропуски: находит самый ранний день, представленный в -ODS и ещё не собранный в DDS, затем при необходимости расширяет до него левую -границу. +Сессии собираются от первого отсутствующего дня до последнего прожитого. +`world_position` хранит номер следующего дня для генерации, поэтому +последний прожитый день равен `min(EventDate) + world_position - 1`. +Начало календаря берется из ODS; события ранних дней там не удаляются по TTL. +Один `max(EventDate)` не годится: события за день есть и во время его +генерации, когда последние визиты еще не закончились. +Точный запрос — [`session_scope.sql`](../../sql/dds/session_scope.sql). -Позиция мира заказам не отвечает: она говорит, какой день прожит, а не какой -доехал. Прогон дня D везёт слепок дня D − 1, да и ставится позиция раньше -трансформации, так что после неудачного приёма обгоняет хранилище. Слой обязан -считать объём по тому, что у него есть, а не по тому, что мир о себе заявляет. +Для заказов диапазон задает последний принятый слепок ODS. Среди строк +с максимальной `snapshot_date` выбираются самая ранняя и самая поздняя +даты оформления заказа. Эти дни еще могут измениться. Если в DDS +отсутствует более ранний день, который есть в ODS, левая граница +расширяется до него. Все дни отрезка пересобираются. -Для сессий границу задаёт `world_position`: она отличает полностью прожитый -день от ещё идущего. Сами события этой границы не показывают: и посреди дня D, -и сразу после его конца их позднейший день равен D. Если собрать сессии посреди -дня, визит, начатый в 23:50, останется обрезанным: завершённый день слой -повторно не собирает. Поэтому последний прожитый день равен первому дню -событий плюс позиция минус один -([`sql/dds/session_scope.sql`](../../sql/dds/session_scope.sql)). -Оговорка про «доехал» остаётся в силе и здесь: за хвост потока отвечает не -позиция, а «Готовность источника». +Позиция мира для выбора дней заказов не подходит: с событиями дня D +приходит слепок за D−1. Объем рассчитывается по принятым строкам ODS; +их наличие еще не доказывает полноту слепка. + +Карта идентичностей пересчитывается целиком. Витрины выбирают дни +по своим источникам — см. [ритм пересборки DM](dm.md#ритм-пересборки). +Поиск отсутствующих дней проверяет только наличие даты: частично +заполненный день требует отдельной проверки. ## Готовность источника -Первый шаг `dds_transform` проходит два барьера за общий срок в 300 секунд. -Сначала клиент `confluent-kafka` сверяет подтвержденные смещения группы -`clickstream_hits` с верхними смещениями обоих разделов `hits`. Группа, оба -раздела и все смещения должны существовать. Суммарное отставание должно стать -нулевым. Каждый опрос пишет в журнал оба смещения и отставание по разделу. +Перед сборкой DDS задача `wait_for_source` последовательно ждет двух +условий: -Только после этого шаг проверяет обе ноды ClickHouse через -`clusterAllReplicas`: у `ods.event_dist` в `system.distribution_queue` не -должно остаться файлов. Каждый опрос пишет в журнал ноды, число файлов и -`last_exception` очереди. Оба барьера опрашиваются раз в три секунды и делят -один крайний срок. После Kafka срок не отсчитывается заново. +1. Группа `clickstream_hits` дочитала оба раздела топика `hits`. + Подтвержденные смещения сравниваются с верхними смещениями Kafka. + Группа и смещения должны существовать, суммарное отставание — равняться нулю. +2. В `system.distribution_queue` у `ods.event_dist` не осталось файлов + доставки на шарды. Запрос через `clusterAllReplicas` читает обе ноды. -Зелёный приём заказов о событиях не говорит ничего — топики разные и -потребители разные. Одного нулевого отставания Kafka тоже мало: чтец мог -подтвердить сообщение раньше, чем Distributed-таблица доставила файл на шард. -Два барьера закрывают обе части пути до того, как сессии выберут свой объем. +Опросы идут с паузой в три секунды; на обе стадии задан общий срок +300 секунд. Журнал показывает смещения и отставание Kafka, затем число +файлов и `last_exception` очереди на каждой ноде. Если готовность не +подтверждена за срок ожидания, задача завершается ошибкой. + +Одного нулевого отставания Kafka мало: сообщение может быть подтверждено, +пока Distributed-таблица еще доставляет его на шард. Успех `orders_ingest` +тоже не заменяет эти проверки: у заказов другой топик и потребитель. + +Эти условия проверяют доставку на момент опроса. Они не останавливают +поступление новых событий и не сверяют содержимое таблиц. Поэтому +`world_position` отдельно ограничивает сборку сессий прожитыми днями, +а качество результата проверяет `dq_check`. ## Идемпотентность -Партиция не заменяется простой вставкой. День собирается в таблицу-двойник — -вставка через Distributed раскладывает строки по шардам, — затем -`REPLACE PARTITION` на кластере атомарно подменяет кусок каждого шарда. Эта -механика — часть самих трансформаций и строится этапом 4: без неё дневной -прогон не повторяем. Сущности, пересчитываемые целиком, идемпотентны и без -замены: полная перезапись в `ReplacingMergeTree` — так живут карта -идентичностей и витрина `daily_traffic` ([слой DM](dm.md)). +Идемпотентность означает, что повторная обработка тех же исходных данных +не добавляет копии деловых строк. Служебные `_load_id` и `_load_ts` +при новой сборке меняются. + +Для дневных таблиц порядок такой: `scope` выбирает даты, `rebuild` очищает +донор и собирает в него весь диапазон, затем отдельные задачи `replace` +публикуют дни через `REPLACE PARTITION`. Донор — служебная таблица с той же +структурой и раскладкой по шардам, что у цели. Запись в `_stage_dist` +с `distributed_foreground_insert = 1` дожидается доставки строк перед заменой. + +Замена выполняется на локальных `_rep` с `ON CLUSTER`. Она атомарна +на каждом шарде: читатель видит старую или новую партицию. Единой +транзакции для двух шардов и всех дней нет. Если замена прошла только +на одном шарде, другой может сохранить прежние данные. Повтор задачи +`replace` с тем же готовым донором повторит замену без добавления копий. + +Новый запуск всего конвейера заново выбирает дни. Он не обязан повторить +все дни неудачного запуска: если дата уже есть в целевой таблице, поиск +пропусков ее не выберет. В частности, изменение SQL сессий само по себе +не заставляет пересчитать заполненные дни. Повторная сборка выбранного +дня разобрана в [задании ступени 4](../labs/4-1-session-pipeline.md). + +Карта идентичностей и `daily_traffic` обновляются иначе: каждый запуск +добавляет полный результат новой версией в `ReplacingMergeTree(_load_ts)`. +Публичные представления читают текущие строки через `FINAL`. Старые версии +могут физически оставаться до фонового слияния; повторяемость относится +к результату чтения, а не к числу строк на диске. ## Чего конвейер о себе не знает diff --git a/docs/architecture/storage.md b/docs/architecture/storage.md index 5dfc360..7d14ed6 100644 --- a/docs/architecture/storage.md +++ b/docs/architecture/storage.md @@ -30,115 +30,92 @@ keeper, Kafka, каркас сервисов. Этап 2 собран: в `sql/d ## Имена объектов -Имя объекта заканчивается тем, что это за объект: +Суффикс показывает назначение объекта: | Суффикс | Что это | |---|---| -| `_rep` | локальная таблица шарда, движок семейства `Replicated*` | -| `_dist` | `Distributed` поверх одноимённой локальной | -| `_kafka` | таблица на движке `Kafka` | -| `_file` | таблица на движке `File` — чтец файла | -| `_mv` | материализованное представление | -| `_v` | обычное представление, публичная поверхность чтения | +| `_rep` | локальная таблица: хранит данные своего шарда, движок семейства `Replicated*` | +| `_dist` | таблица `Distributed`: читает данные шардов и распределяет запись между ними | +| `_kafka` | таблица на движке `Kafka`: читает сообщения топика | +| `_file` | таблица на движке `File`: читает файл | +| `_mv` | материализованное представление: обрабатывает вставляемые данные | +| `_v` | обычное представление: выполняет сохраненный запрос при чтении | -Суффикс носит каждый физический объект. Голого имени у таблицы не существует: -запрос к `stg.hits_raw` даёт громкую ошибку «нет такой таблицы» — а под голым -именем в документах и разговоре понимается сущность, у которой этих объектов -несколько. Единственное исключение — словари: у них воплощение одно, шардировать -нечего, и суффикс ничего не различал бы. В зоне `dic` обе формы стоят рядом и -правило видно целиком: `dic.products_file` — чтец файла, `dic.products` — сам -словарь. +Для запросов к моделям DDS и витринам DM используйте `_v`. Представления +явно перечисляют колонки, поэтому добавление поля в физическую таблицу +не меняет набор полей для читателя. У `dds.event_v` нет отдельной таблицы +DDS: представление читает события ODS. -Распространённая конвенция, где голое имя означает локальную таблицу, а -распределённая получает суффикс `_all`, ошибается иначе: забытый суффикс тихо -возвращает данные одного шарда. Правило спеки «проверки и контрольные суммы — -только по `Distributed`» такую тишину переживает плохо. +В документации имя без суффикса, например `stg.hits_raw`, обозначает +сущность с несколькими объектами. Таблицы с таким именем нет: в запросе +нужно указать суффикс. Исключение — словари ClickHouse: `dic.products` +обращается к каталогу товаров, а `dic.products_file` читает его файл. -Второе свойство — в списке по алфавиту объекты группируются по сущности, а не по -технологии: все четыре объекта топика `hits` стоят рядом, потому что различаются -хвостом, а не началом имени. Речь про дерево в клиенте и про `ORDER BY name`: -порядок выдачи `SHOW TABLES` документация не оговаривает. +В STG имя берется из топика: `hits_raw`, `orders_raw`. В типизированных +слоях оно называет сущность в единственном числе: `event`, `order`, +`session`. Объекты одной сущности различаются суффиксом и стоят рядом +при сортировке имен по алфавиту. -Начало имени — сущность, и берётся она в разных слоях из разных мест. В STG имя -приходит от транспорта: слой хранит то, что доехало по топику, и зовётся именем -топика — `hits_raw` от `hits`, `orders_raw` от `orders`. В типизированных слоях -имя приходит от предметной области и стоит в единственном числе: `event`, -`order`, `session`. Граница между «как привезли» и «что это такое» -проходит по STG, и имена её показывают. - -В DDS публичный договор сущности заканчивается на `_v`. Потребительские запросы -и витрины называют только представление; `_rep` и `_dist` служат загрузке, -диагностике и обслуживанию. Представление перечисляет колонки явно, даже если -его запрос — простая проекция над `_dist`: договор не выдаёт устройство -кластера и случайно добавленные в физическую таблицу колонки. У события -физической копии в DDS нет, но правило то же: `dds.event_v` — его публичный -договор. - -Право доступа здесь намеренно шире договора. Роль `bi_reader` может читать -любые объекты ODS, DDS и DM, но физические объекты от этого не становятся -публичной поверхностью. Конвенцию держат явный список колонок, документация и -ревью, а не запрет на физические таблицы. - -Широкое право принято в [ADR 0007](../adr/0007-clickhouse-access.md): узкая -граница поверх `Distributed` потребовала бы отдельной механики. Механизм -ClickHouse и отклонённые варианты записаны в +`_rep` и `_dist` нужны для загрузки, диагностики и обслуживания. Роль +`bi_reader` может читать их в ODS, DDS и DM, но для обычного анализа +принято использовать `_v`. Это соглашение о чтении, а не ограничение прав; +причина широкого доступа описана в +[ADR 0007](../adr/0007-clickhouse-access.md) и [исследовании](../research/2026-08-25-clickhouse-distributed-view-security.md). ## Раскладка по шардам -Пишем только в `_dist`. Локальные таблицы остаются для чтения и обслуживания — -операций с партициями, ручной переобработки. Правило не про удобство: при записи -через распределённую таблицу раскладку определяет ключ шардирования, то есть -свойство данных, а при записи в локальную — то, какая нода случайно выполняла -код. Отсюда урок стенда: какая нода читала топик, меняется между прогонами -(видно в колонке `consumer_host`), а куда легли данные — нет. +Шард хранит часть строк. При записи через `_dist` место строки выбирает +ключ шардирования. При прямой записи в `_rep` она остается на той ноде, +где выполнена вставка. Поэтому данные записывают через `_dist`, +а операции с партициями выполняют на локальных `_rep`. -Ключи шардирования: `cityHash64(ClientID)` у событий, `cityHash64(order_id)` у -заказов, `cityHash64` сырой строки у STG и у таблицы ошибок. У первых двух хеш -выбран против перекоса: структурированный числовой идентификатор распределяется -по остатку от деления неравномерно. У сырого слоя выбора нет — строку иначе не -разложишь; там хеш даёт другое свойство, одинаковые сообщения ложатся на один -шард. +У событий ключ — `cityHash64(ClientID)`, у заказов — +`cityHash64(order_id)`, у STG и таблиц ошибок — хеш сырой строки. +Хеш распределяет числовые идентификаторы по шардам и направляет одинаковые +сообщения на один шард. У сырой строки и разобранного события разные +ключи: они могут оказаться на разных шардах. Счетчики STG и ODS сравнивают +по всему кластеру через `_dist`, а не отдельно по каждой ноде. -Ключи у слоёв разные, и это имеет наблюдаемое следствие: сырая строка и -разобранное из неё событие почти всегда оказываются на разных шардах. -Пошардовые счётчики STG и ODS поэтому не сходятся и сходиться не должны — -сверять слои можно только через `_dist`. +При `INSERT INTO цель_dist SELECT … FROM источник_dist` важна настройка +`parallel_distributed_insert_select`. При значении `2` и одном кластере +вставка может выполняться локально на шардах, сохраняя раскладку источника +вместо ключа цели. Если ключи различаются, укажите +`parallel_distributed_insert_select = 0`: нода, принявшая запрос, +соберет результат и распределит его по ключу цели. Так сделано в +[`sql/ods/order_parse.sql`](../../sql/ods/order_parse.sql) и при сборке +выручки. Неверную раскладку проверяют по `hostName()` и ожидаемому шарду +строки; общий счет по `_dist` ее не показывает. -Разные ключи у слоёв ставят капкан на вставке `INSERT INTO цель_dist SELECT … -FROM источник_dist`. Заводское значение `parallel_distributed_insert_select` — -2: если обе таблицы на одном кластере, вставка исполняется локально на каждом -шарде, минуя ключ шардирования цели. Строки наследуют раскладку источника, и -ошибки при этом нет — расхождение видно только пересчётом ключей по -`hostName()`. Так половина заказов и легла на оба шарда, где -`ReplacingMergeTree` через границу шарда дубли не схлопывает (замер — «Что -проверено», #137). Лекарство — явное `parallel_distributed_insert_select = 0` -в `SETTINGS` такой вставки: раскладку делает инициатор, по ключу цели -(`sql/ods/order_parse.sql`). Капкан ждёт и сборки DDS: они тоже будут читать -`_dist` и писать в `_dist` с другим ключом. +События, `dds.session_dist` и `dds.identity_map_dist` распределены +по `cityHash64(client_id)` (в ODS поле называется `ClientID`). Данные +одного посетителя лежат вместе — это ко-локация. Заказы DDS и их сверка +сохраняют раскладку по `cityHash64(order_id)`. Обычное соединение допускается +по ключу ко-локации; для остальных соединений применяется `GLOBAL`. +Само совпадение ключей еще не гарантирует локального выполнения запроса +через представления — пример есть в [описании `daily_traffic`](dm.md#dmdaily_traffic_v). -Ключи ко-локации названы заранее, потому что на них стоит политика соединений из -раздела 6 спеки: обычное соединение разрешено только по ключу ко-локации, всё -прочее — через `GLOBAL`. Значит `dds.session_dist` и `dds.identity_map_dist` -шардируются по `cityHash64(client_id)`. Объекты DDS с зерном заказа сохраняют -ко-локацию по `cityHash64(order_id)`; ключи остальных частей модели и витрин -появляются вместе с ними. - -Замена партиций не используется для `ods.order`: версии прошлых дней -доливаются, а прямое чтение Kafka не задаёт границы полного слепка ([ADR -0010](../adr/0010-order-versions-in-ods.md)). Если партиционная пересборка -понадобится будущим объектам DDS или DM, партиция-донор должна быть заранее -разложена по шардам по тому же ключу. Форму донора следует решать вместе с таким -объектом, а не переносить на ODS заранее. - -Первым такой донор понадобился заказу DDS, и правило на нём видно целиком: -`dds.order_stage_rep` объявлен через `AS dds.order_rep` и шардируется тем же -`cityHash64(order_id)`. Совпадение структуры требует сама операция, а совпадение -ключа — устройство замены: она идёт по локальным таблицам, шард за шардом, и -строка, попавшая в доноре не на свой шард, при подмене пропала бы с одного шарда -и задвоилась на другом. Доводы целиком — в +Для дневной пересборки 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 показан в [`sql/ddl/45-dds-tables.sql`](../../sql/ddl/45-dds-tables.sql). +`REPLACE PARTITION` копирует готовую партицию между локальными таблицами +на каждом шарде; заново распределять строки по ключу цели эта операция +не будет. Поэтому донор должен сразу иметь ту же раскладку, что и цель. +Перед заменой вставка с `distributed_foreground_insert = 1` дожидается +доставки данных на шарды. + +В `ods.order` этот способ не применяется: туда добавляют версии заказов +из слепков, а актуальные строки выбираются при чтении +([ADR 0010](../adr/0010-order-versions-in-ods.md)). + ## Служебные колонки Собственные колонки не повторяют имён виртуальных. Виртуальные даёт движок: diff --git a/sql/dds/session_rebuild.sql b/sql/dds/session_rebuild.sql index daef9d2..3939858 100644 --- a/sql/dds/session_rebuild.sql +++ b/sql/dds/session_rebuild.sql @@ -1,35 +1,17 @@ --- DDS: нарезка событий на визиты и сборка дней в таблицу-двойник. +-- DDS: сборка сессий выбранных дней в донор dds.session_stage_dist. +-- Донор очищается целиком; одна вставка собирает весь диапазон. -- --- Двойник очищается целиком в начале, дни собираются одной вставкой на весь --- отрезок — по правилу заказа (sql/dds/order_rebuild.sql). +-- Первое окно помечает событие единицей, если с предыдущего события того же +-- посетителя за тот же день прошло больше 1800 секунд. Второе окно суммирует +-- эти единицы: получается номер визита внутри посетителя и дня. +-- PARTITION BY (client_id, event_date) разделяет визиты по модельной полночи. +-- Учитываются все типы событий. -- --- Нарезка — центральный расчёт слоя, и стоит она на двух окнах подряд. --- Первое спрашивает у каждого события, сколько прошло с предыдущего события --- той же куки в тот же день, и ставит единицу там, где пауза длиннее --- таймаута: это шов между визитами. Второе считает нарастающий итог этих --- единиц — номер визита внутри дня куки. Дальше визит становится обычной --- группировкой. --- --- Границу суток окна не проверяют: она уже в разбиении PARTITION BY --- (client_id, event_date), поэтому визит через полночь не склеится и при --- короткой паузе. Так правило визита ложится в запрос целиком: одна кука, --- один день, паузы не длиннее таймаута (CONTEXT.md, «Визит»). --- --- Первое событие куки за день сравнивается само с собой — третий аргумент --- lagInFrame. Паузы у него нет, шов ему не нужен, и номер первого визита --- получается нулём: номера различают визиты внутри дня куки, а не считают их --- от единицы. --- --- Порядок в окнах полный — метка времени и event_id. Окна считаются двумя --- разными сортировками (видно в EXPLAIN PIPELINE), и при равных метках они --- вправе разойтись: шов достался бы одной строке, а нарастающий итог другой, --- и сессия разъехалась бы надвое. --- --- Тридцать минут написаны числом, а не приехали параметром: это правило --- аналитики, а не константа мира. Генератор режет визиты тем же порогом; --- разойдись эти два числа — увидит счётная сверка с VisitID --- (docs/architecture/dds/session.md, «Сверка с эталоном»). - +-- У первого события lagInFrame возвращает его собственное время: разница +-- равна нулю, opens_visit = 0, и первый visit_no тоже равен нулю. +-- Оба окна используют (event_time, event_id), чтобы при равном времени +-- признак начала и накопленный номер визита относились к одним событиям. +-- Порог 1800 секунд — правило визита; он не передается параметром. ALTER TABLE dds.session_stage_rep ON CLUSTER clickstream_cluster DROP PARTITION ALL; INSERT INTO dds.session_stage_dist @@ -93,8 +75,8 @@ WITH FROM seams ) SELECT - -- Чекан визита: кука и его начало. Время входит в хеш секундой эпохи, - -- поэтому id держится за содержание сессии, а не за порядок сборки. + -- Идентификатор зависит от посетителя и начала визита. + -- Повторная сборка тех же событий дает тот же session_id. cityHash64(client_id, min(event_time)), client_id, event_date, @@ -107,8 +89,10 @@ SELECT countIf(event_type = 'purchase'), argMin(url, event_time), argMax(url, event_time), - -- Атрибуция и паспорт куки берутся с первого события визита. У паспорта - -- любое событие дало бы то же: устройство и город у куки постоянны. + -- UTM, устройство и город берутся по минимальному event_time. + -- Как и у страниц выше, при равном времени выбор между событиями + -- не определен: event_id в эти argMin/argMax не входит. + -- Устройство и город на стенде постоянны для одной куки. argMin(utm_source, event_time), argMin(utm_medium, event_time), argMin(utm_campaign, event_time), @@ -120,12 +104,7 @@ SELECT now64(3, 'UTC') FROM numbered GROUP BY client_id, event_date, visit_no --- Капкана parallel_distributed_insert_select здесь нет, и это выигрыш --- ко-локации: ключ шардирования у источника и цели один — cityHash64 от --- куки, — поэтому визит ляжет туда же, где лежали его события, исполнись --- вставка хоть на инициаторе, хоть локально на шардах. Заказу приходится --- называть ноль вслух: совпадение ключей у него случайное --- (sql/dds/order_rebuild.sql). Здесь выбор сделан обратный и он тоже --- осознанный: настройку не называть, чтобы не притворяться, будто её значение --- что-то решает. +-- Источник и цель распределены по cityHash64(client_id): локальная вставка +-- сохраняет нужную раскладку, parallel_distributed_insert_select менять +-- не требуется. Перед заменой партиций ждем доставки строк на шарды. SETTINGS distributed_foreground_insert = 1; diff --git a/sql/dds/session_replace.sql b/sql/dds/session_replace.sql index 6b220ba..b854ef5 100644 --- a/sql/dds/session_replace.sql +++ b/sql/dds/session_replace.sql @@ -1,11 +1,10 @@ --- DDS: подмена одного дня сессий собранной партицией двойника. +-- DDS: замена одного дня в session_rep готовой партицией session_stage_rep. +-- ON CLUSTER выполняет замену на локальных таблицах каждого шарда. +-- Замена атомарна на одном шарде, общей транзакции для кластера нет. +-- Повтор с тем же донором не добавляет копий строк. -- --- Механика и доводы целиком — sql/dds/order_replace.sql: замена идёт по --- локальным таблицам, шард за шардом, атомарность пошардовая, день --- подставляет Jinja, потому что параметров ALTER … PARTITION не берёт. --- --- Разница одна, и она про отказ. Несобранный день заказа вернуло бы в работу --- следующее окно; сессиям возвращать нечем, и упавшая замена оставляет дыру --- до тех пор, пока её не найдёт счёт хвоста (sql/dds/session_scope.sql). +-- Дату из session_scope.sql подставляет Jinja: параметр ClickHouse в +-- ALTER ... PARTITION здесь не поддерживается. Поиск дней следующего +-- запуска не обнаружит неполный день, если часть его строк уже есть в DDS. ALTER TABLE dds.session_rep ON CLUSTER clickstream_cluster REPLACE PARTITION '{{ params.day }}' FROM dds.session_stage_rep; diff --git a/sql/dds/session_scope.sql b/sql/dds/session_scope.sql index f37425d..37f3b7b 100644 --- a/sql/dds/session_scope.sql +++ b/sql/dds/session_scope.sql @@ -1,50 +1,30 @@ --- DDS: объём пересборки сессий — какие дни собирает этот запуск. +-- DDS: дни для сборки сессий, от first_day до last_day включительно. +-- Пустой ответ означает пустой ODS; NULL в first_day — отсутствие работы. -- --- Ответ — две даты, границы непрерывного отрезка; список дней между ними --- разворачивает даг. Пустой ответ означает пустой источник. Пустой первый --- день — что собирать нечего: все прожитые дни уже в слое, и прогон честно --- пропускает сборку. +-- position — номер следующего дня для генерации. Последний прожитый день +-- равен min(EventDate) + position - 1. В отличие от max(EventDate), эта +-- граница исключает еще идущий день. Ранние события ODS не удаляются по TTL, +-- поэтому min(EventDate) сохраняет начало модельного календаря. -- --- Файл кончается без точки с запятой: почему — sql/dds/order_scope.sql. +-- До этого запроса даг ждет чтения hits и опустошения очереди ods.event_dist. +-- Позиция мира сама по себе не подтверждает доставку событий. -- --- Правый край — последний прожитый день, и знает его только позиция мира. --- События живой день от прожитого не отличают: и посреди дня D, и сразу после --- его конца max(EventDate) = D. А разница несущая — визит, начатый в 23:50, --- посреди дня ещё идёт, и собранная сессия оказалась бы обрезанной навсегда: --- пересобирать замёрзший день слой не станет --- (docs/architecture/dds/session.md, «Движок и наполнение»). --- --- Позиция — номер первого несыгранного дня, счётчик прожитого. Своей копии --- начала оси стенд из-за неё не заводит: начало он спрашивает у данных, и --- дата дня N в них — константа, начало оси плюс N. Держится якорь на том, что --- ранние дни из ods.event не исчезают: срок жизни назначен сырым данным и таблицам --- ошибок, самому событию — нет (sql/ddl/20-ods-tables.sql). --- --- За «доехал» позиция не отвечает — только за «прожит». Перед расчётом объёма --- первый шаг дага дожидается хвоста потока и очереди доставки: иначе --- недожеванный хвост заморозил бы недособранный день --- (docs/architecture/etl.md, «Готовность источника»). +-- Точки с запятой в конце нет: драйвер добавляет FORMAT к этому запросу. WITH (SELECT min(EventDate) FROM ods.event_dist) AS first_source_day, first_source_day + ({position:UInt16} - 1) AS last_lived_day, - -- Хвост считается от первой дыры, а не от последнего собранного дня: - -- замены партиций падают порознь, и счёт по максимуму собранного дня - -- дыру позади себя не заметил бы (доводы целиком — sql/dds/order_scope.sql). - -- У сессий это единственный способ дыру найти: дышащего окна, которое - -- пересобрало бы день заодно, здесь нет. - -- - -- GLOBAL обязателен: обе таблицы распределённые, и обычный NOT IN сверял - -- бы дни источника с локальной частью слоя на каждом шарде. - -- minOrNull отличает «дыр нет» от даты по умолчанию. + -- Ищем первый день ODS, которого нет в DDS: максимум собранного дня + -- пропустил бы более ранний пробел. Частично заполненный день так не найти. + -- GLOBAL сравнивает со списком дней всего DDS, а не одного шарда. + -- minOrNull возвращает NULL, если пропусков нет. ( SELECT minOrNull(day) FROM (SELECT DISTINCT EventDate AS day FROM ods.event_dist) WHERE day GLOBAL NOT IN (SELECT session_date FROM dds.session_dist) ) AS first_missing_day SELECT - -- Живой день в слое отсутствует всегда, и работой это не считается: - -- дырой он станет, только когда мир его проживёт. + -- Отсутствующий день попадет в сборку только после его завершения. if(first_missing_day <= last_lived_day, first_missing_day, NULL) AS first_day, last_lived_day AS last_day --- FINAL не нужен: повтор доставки не двигает ни край оси, ни набор дней. +-- FINAL не нужен: повтор доставки не меняет начало календаря и набор дат. WHERE (SELECT count() FROM ods.event_dist) > 0 diff --git a/sql/dm/revenue_daily_rebuild.sql b/sql/dm/revenue_daily_rebuild.sql index 46f4e24..e762c7c 100644 --- a/sql/dm/revenue_daily_rebuild.sql +++ b/sql/dm/revenue_daily_rebuild.sql @@ -1,7 +1,8 @@ --- DM: сборка выручки в донор перед заменой дневных партиций. --- DROP PARTITION ALL очищает локальные таблицы на каждом узле; причина не --- применять TRUNCATE разобрана в соседнем sql/dds/order_rebuild.sql. - +-- DM: сборка выручки за выбранный диапазон в донор. +-- DROP PARTITION ALL очищает донор на каждом узле. Вставка ниже дает +-- одну строку на день заказа и категорию товара: только оплаченные заказы. +-- ARRAY JOIN разворачивает согласованные позиции трех массивов вместе. +-- Причина выбора DROP вместо TRUNCATE — в sql/dds/order_rebuild.sql. ALTER TABLE dm.revenue_daily_stage_rep ON CLUSTER clickstream_cluster DROP PARTITION ALL; @@ -22,8 +23,8 @@ SELECT orders, units, CAST(revenue, 'Decimal(18, 2)'), - -- Обычное деление Decimal усекает доли копейки вместо округления; - -- запас масштаба сохраняет их до денежного результата. + -- Делим с шестью знаками после запятой, затем округляем до копеек. + -- Так дробная часть сохраняется до округления среднего чека. CAST( round(divideDecimal(revenue, toDecimal128(orders, 0), 6), 2), 'Decimal(18, 2)' @@ -49,7 +50,8 @@ FROM report_date, product_category ) --- Источник лежит по заказу, цель — по категории. Ноль заставляет инициатор --- слить частичные агрегаты и разложить готовые строки по ключу цели. +-- Источник распределен по заказу, цель — по категории. Значение 0 оставляет +-- итоговую агрегацию и распределение по ключу цели на ноде, принявшей запрос. +-- distributed_foreground_insert = 1 ждет доставки строк перед заменой. SETTINGS distributed_foreground_insert = 1, parallel_distributed_insert_select = 0; diff --git a/sql/dm/revenue_daily_replace.sql b/sql/dm/revenue_daily_replace.sql index 0fedd96..29d98bc 100644 --- a/sql/dm/revenue_daily_replace.sql +++ b/sql/dm/revenue_daily_replace.sql @@ -1,8 +1,8 @@ --- DM: подмена одного дня выручки готовой партицией донора. --- --- День приходит из результата revenue_daily_scope.sql, а не от пользователя. --- Jinja нужна потому, что ClickHouse не подставляет параметр в PARTITION. --- Замена идет на локальной таблице ON CLUSTER: граница атомарности и повтор --- разобраны на таком же шаге в sql/dds/order_replace.sql. +-- DM: замена дня выручки готовой партицией донора. +-- Дату из 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 c15154f..ca6ec8d 100644 --- a/sql/dm/revenue_daily_scope.sql +++ b/sql/dm/revenue_daily_scope.sql @@ -1,14 +1,11 @@ --- DM: объем пересборки выручки. +-- DM: диапазон пересборки выручки, first_day и last_day включительно. +-- Последний слепок ODS задает даты оформления еще изменяемых заказов. +-- Агрегат читает DDS; ODS нужен только для границ диапазона. +-- Левая граница расширяется до первого дня с оплаченными заказами DDS, +-- которого нет в витрине. Так первый запуск охватывает историю выручки. +-- По наличию даты нельзя обнаружить частично заполненный день. -- --- Последний слепок ODS называет дни, состояние которых источник еще может --- изменить. Сам агрегат читает только DDS; ODS здесь нужен лишь для границ. --- Первый запуск и повтор после частичного отказа расширяют левую границу до --- первого дня оплаченных заказов, которого еще нет в витрине. --- Почему хвост начинается с первого пропуска и чем за это платит ODS, --- разобрано в sql/dds/order_scope.sql. --- --- Файл кончается без точки с запятой: даг читает его одним запросом и ждет --- одну строку с границами. +-- Точки с запятой в конце нет: драйвер добавляет FORMAT к этому запросу. WITH (SELECT max(snapshot_date) FROM ods.order_dist) AS last_snapshot, ( @@ -24,8 +21,7 @@ WITH ( SELECT minOrNull(report_date) FROM (SELECT DISTINCT order_date AS report_date FROM dds.order_v WHERE status = 'paid') - -- Правая сторона тоже распределенная: GLOBAL собирает множество дней - -- на инициаторе и не превращает запрос в запрещенное двойное чтение. + -- GLOBAL собирает список дней витрины со всех шардов. WHERE report_date GLOBAL NOT IN ( SELECT report_date FROM dm.revenue_daily_dist ) @@ -33,7 +29,7 @@ WITH SELECT least(window_start, coalesce(first_missing_day, window_start)) AS first_day, window_end AS last_day --- Без ODS неизвестно окно. Без готового DDS донор останется пустым, а замена --- сотрет уже собранные дни выручки. Поэтому для прогона нужны оба источника. +-- Для выбора диапазона нужны заказы в ODS, для расчета — в DDS. +-- Эта проверка наличия строк не подтверждает полноту источников. WHERE (SELECT count() FROM dds.order_v) > 0 AND (SELECT count() FROM ods.order_dist) > 0