Compare commits

Author SHA1 Message Date
ddadmin 086007fbd4 docs(warehouse): объяснены расчет строк и замена дневных разделов
Зачем: выражения «собираются в донор» и «публикуются дни» требовали
знать устройство конвейера до чтения объяснения.

Что: описаны назначение промежуточных таблиц и последовательность
расчета, записи и замены разделов. Перечитаны остальные объяснения:
выбор дней, разворачивание позиций заказа, проверка сессий и повтор запуска.

Проверка: make lint, git diff --check, локальные ссылки и якоря.
SQL без комментариев и AST обоих дагов без строк документации не изменились.

Refs: #249
2026-09-13 15:16:26 -04:00
ddadmin 5f317ba8d2 docs(warehouse): упрощены объяснения витрин, сессий и пересчета
Зачем: читателю ступеней 1–4 нужно проследить выбор дней, расчет строки
и публикацию результата без расшифровки метафор.

Что: согласованы справочники DM, сессий, ETL и два раздела хранилища
с комментариями шести SQL-примеров. Уточнены ограничения повторного
запуска и выбора атрибутов при равном времени событий.

Проверка: make lint, git diff --check, локальные Markdown-ссылки и якоря.
SQL без комментариев совпадает; AST Python без строк документации совпадает.
Исторические разделы и материалы вне границ задачи сохранены.

Refs: #249
2026-09-13 14:33:39 -04:00
ddmitry 84dc718f62 docs(context): влита редактура словаря (#252) 2026-09-13 21:00:12 +03:00
ddadmin 5a3e3f92d9 docs(context): упрощен словарь понятий для начинающего
Зачем: понимать термины стенда без изучения устройства генератора.
Что: понятия сгруппированы по учебным темам, определения упрощены;
уточнены идентификаторы, модельное время и повторный запуск стенда.
Проверка: сохранены все 43 прежних термина; проверены 11 входящих
и 6 внутренних ссылок, якорь «Язык» и git diff --check.
2026-09-13 13:59:37 -04:00
13 changed files with 798 additions and 752 deletions

No files matched your search

+247 -201
View File
@@ -1,229 +1,275 @@
# Кликстрим-платформа (стенд v2) # Кликстрим-платформа (стенд v2)
Словарь понятий проекта: одни и те же слова для одних и тех же вещей — Словарь понятий, которые встречаются в учебных заданиях, документации и коде
у владельца, кода, документов и агентов. Только язык, никаких решений. стенда.
## Язык ## Язык
**Хранилище**: - [Люди и события](#люди-и-события)
Аналитическая база стенда — кластер ClickHouse со слоями STG/ODS/DDS/DM. - [Хранилище и форматы данных](#хранилище-и-форматы-данных)
Принимающая сторона границы «трекер | хранилище»: нормализует имена и стили - [Заказы](#заказы)
источников, строится по их документации. - [Модельное время и запуск стенда](#модельное-время-и-запуск-стенда)
_Избегать_: склад, склад данных - [Устройство генератора](#устройство-генератора) — для чтения его кода
и технических справочников.
**Сырые данные**: ### Люди и события
Слой STG хранилища и его содержимое: сообщение топика, положенное строкой,
как пришло, без разбора. В прозе — сырые данные и сырой слой; про
конкретную таблицу или базу — STG.
_Избегать_: сырьё
**Состав мира**:
Постоянная часть мира генератора — популяция посетителей, их привычки,
двухкуковые пары. По дням его выдаёт план состава.
_Избегать_: состояние мира
**План состава**:
Способ спросить состав мира: функция зерна, выдающая его по дням —
когорту новых кук, их возвраты, назначенные заказы двухкуковых пар.
**Подпоток**:
Ветвь дерева случайности генератора: своё зерно у состава мира, у каждого
дня и у каждого компонента дня. Подпоток задан позицией в дереве, а не
порядком вычислений.
**Конфигурация мира**:
Модуль чистых данных со всеми числами мира: приток, профили возвратов, доли
покупателей, D0. Правка модуля — смена мира. Модуль-близнец контракта схемы:
там колонки, здесь числа.
**Когорта дня**:
Люди, впервые пришедшие в мир в один день, со всеми их куками и днями
активности. Единица плана состава: когорта — функция зерна и номера дня.
**Человек**: **Человек**:
Устойчивая личность в составе мира, которой принадлежат одна или две куки. Участник модели магазина, который может заходить с одного или двух
Не поле источника: кликстрим видит куку, а заказ — пользователя магазина. устройств. Генератор знает, какие идентификаторы посетителя ему принадлежат,
но в событиях сайта этой связи нет.
**Посетитель**: **Посетитель**:
Кука, наблюдаемая в кликстриме и обозначенная `ClientID`; единица Браузер на устройстве, который счетчик сайта узнает по куке — сохраненному
`uniq(ClientID)`. Один человек может быть представлен несколькими посетителями. в браузере идентификатору. В кликстриме он обозначается `ClientID`:
_Избегать_: «посетитель» про человека после склейки `uniq(ClientID)` считает посетителей, и у одного человека их может быть несколько.
_Избегать_: «посетитель» про человека после объединения его идентификаторов.
**Пользователь магазина**: **Пользователь магазина**:
Представление человека со стороны заказов; в записи заказа обозначается Человек, известный серверной части магазина под идентификатором `user_id`.
`user_id`. Через заказы связывает разные куки человека, в кликстрим напрямую Он указан в заказах и позволяет связать разные `ClientID` одного человека;
не попадает. в кликстриме поля `user_id` нет. Например, покупки с телефона и ноутбука
могут иметь разные `ClientID`, но один `user_id` в заказах.
**Приток**:
Появление новых кук на всём протяжении оси модельного времени; единица —
кука (`ClientID`). Из-за притока накопленная аудитория растёт с
горизонтом и не совпадает с дневной.
**Хвост возвратов**:
Окно активности, отсчитанное от первого дня человека и общее на обе
его куки; дольше окна кука не возвращается.
**Предыстория**:
Когорты плана с первым днём активности до D0; событий не порождают.
**День активности**:
День, в который кука хоть раз появилась в мире: день её рождения или день
возврата. Единица плана состава: за день он у куки один, а визитов внутри
него бывает несколько.
_Избегать_: визит (визит — про сессию)
**Визит**: **Визит**:
Подряд идущие события одной куки без пауз длиннее 30 минут, не пересекающие Последовательность событий одного посетителя без пауз длиннее 30 минут
границу суток. Поле `VisitID` — эталон для лабы сессий. «Сессия» — то же и без перехода через полночь по часовому поясу счетчика. Сессия означает
понятие словами аналитики: в данных и в коде оно зовётся визитом, но имена то же самое; поле `VisitID` служит эталоном в задании по нарезке сессий.
вроде «лаба сессий» и «сборка сессий» остаются.
**День активности**:
День, в который посетитель появился на сайте хотя бы раз: впервые или
вернувшись. За один день активности у него может быть несколько визитов.
_Избегать_: визит в значении целого дня активности.
**Паспорт куки**: **Паспорт куки**:
Устройство и город, приписанные куке на всю жизнь: кука — это браузер на Устройство и город, закрепленные за посетителем во все дни его активности.
устройстве. Держит их план состава, разворачивает в поля события Генератор записывает их в поля событий. У двух посетителей одного человека
день-функция; у двухкуковой пары город один на две куки, устройства разные. город совпадает, а устройства различаются.
**Покупатель**:
Человек, которого план состава пометил склонным покупать. Метка значима:
помеченный доходит до заказа заметно чаще прочих, но и непомеченный иногда
покупает. Из покупателей отбираются двухкуковые пары.
_Избегать_: «покупатель» про того, кто купил в конкретный день — это визит
с заказом.
**День-функция**:
Функция (зерно, номер дня), выдающая упорядоченный поток событий этих
модельных суток. Состояния между днями нет: день D не зависит от того,
прожиты ли дни до него.
**Торговое событие**: **Торговое событие**:
Событие корзины или покупки — отдельная строка потока, а не просмотр Запись о добавлении товара в корзину или о покупке: `add_to_cart` или
страницы. Садится на ту страницу, где случилось: корзина — на карточку `purchase`. Это отдельная строка кликстрима со страницей, где произошло
товара, покупка — на страницу подтверждения заказа. действие: карточкой товара для корзины или страницей подтверждения для покупки.
### Хранилище и форматы данных
**Хранилище**:
Аналитическая база стенда в ClickHouse. Данные проходят слои STG, ODS, DDS
и DM: от исходных сообщений до витрин для отчетов. Хранилище принимает
данные по документации источников и приводит их к своей модели и именам.
_Избегать_: склад, склад данных.
**Сырые данные**:
Исходные сообщения Kafka, сохраненные в слое STG без разбора: одно сообщение
занимает одну строку таблицы. Слой STG также называют сырым слоем.
_Избегать_: сырьё.
**Справочник**: **Справочник**:
Данные, которые хранилище не производит, а получает готовыми со стороны: Готовые сведения из внешнего источника, которыми хранилище дополняет
содержимым владеет источник вне хранилища, слои только читают. Живут в базе свои данные. На стенде они находятся в базе `dic`; пока это только каталог
`dic`. Первый и пока единственный — каталог товаров. товаров. Объект `DICTIONARY` ClickHouse — одна из форм доступа к справочнику.
_Избегать_: словарь — этим словом в проекте зовут и глоссарий понятий, и _Избегать_: словарь в значении справочных данных: это слово уже обозначает
объект `DICTIONARY` ClickHouse, одну из форм, в которой справочник доступен. `DICTIONARY` и словарь понятий.
**Каталог товаров**: **Каталог товаров**:
`data/catalog/products.csv` — общий файл генератора и словаря Список товаров с артикулами, названиями, категориями, брендами, ценами
ClickHouse. Форма файла решена, длина — нет: строки дописываются. и уровнями спроса. Файл `data/catalog/products.csv` общий для генератора
и справочника ClickHouse; товары можно добавлять строками в этом формате.
**Уровень спроса**:
Свойство товара в каталоге: как часто открытую карточку кладут в корзину.
Три значения — магнит, обычный, залёживается. Постоянная часть мира, а не
поведение дня: он и делает конверсию «просмотр → корзина» по товарам и
брендам различимой.
**Ось модельного времени**:
Собственный календарь мира генератора. Дни считаются от фиксированного
первого дня D0; реальный календарь в модели не участвует. Между прогонами
живут только зерно и позиция на оси.
**Стартовый мир**:
Первые восемь дней оси (понедельник по понедельник), которыми стенд
наполняется при каждом `make up`. Он же эталонный мир, всегда один и тот же:
на нем принимаются следующие этапы.
_Избегать_: зерновой мир
**Паспорт мира**:
`data/world-inventory.json` — единственное, что о мире хранится в git: шапка
(зерно, версия генератора, хеш каталога), строка на каждый день событий
(число и хеш байтов) и строка на каждый отправленный слепок (хеш байтов).
Счётчик сверх этого заводится только с названным читателем. Сам мир не
хранится, а пересчитывается; паспорт отвечает на один вопрос: тот ли это мир.
_Избегать_: опись мира, манифест, мини-манифест
**Пошаговый режим**:
Базовый способ движения по оси модельного времени: «прожить следующий
день» — явное действие.
**Живой день**:
Проигрывание текущего модельного дня в реальном времени с ускорением;
включается по требованию — постоянным фоном идёт, только пока включён
выключатель `world_live`.
**Пакетный режим**:
Проигрывание готового дня пачкой, без темпа: заливка снимка при старте
стенда, пересборки и проверки.
**Даги модельного времени**:
Даги Airflow, которыми двигают ось модельного времени: шаговые даги играют
день — пачкой или в темпе, — а выключатель держит стенд живущим, пока включён.
Коротко — даги времени.
_Избегать_: пульт мира, управляющие даги
**Шаговый даг**:
Даг модельного времени, который играет день и ставит позицию на оси:
`world_next_day` — пачкой, `world_live_day` — в темпе. Запускается руками,
расписания и паузы не имеет; автоматику поверх него даёт выключатель.
_Избегать_: работник
**Граница суток**:
Единственный структурный шов модели: сессии режутся по ней, слепок заказов
снимается на ней, день проживается только целиком.
**Контракт схемы**:
Python-модуль с описателями колонок события — собственность генератора.
Из него выводятся генератор, валидация и документация формата; хранилище
строится по документации, не по модулю.
**Нормализованное имя**: **Нормализованное имя**:
Имя колонки источника, приведённое к нашему стилю (snake_case). Живёт в Имя поля источника, записанное в стиле `snake_case`: например,
контракте схемы и в описании выгрузки. Не то же, что имя атрибута в модели `ClientID` становится `client_id`. Оно указано в описании выгрузки и
данных: слой DDS складывает модель и называет атрибуты по ней. контракте схемы. Слой DDS выбирает имена по своей модели данных, поэтому
они не обязаны повторять нормализованные имена источника.
**Описание выгрузки**: **Описание выгрузки**:
Публичная документация формата события: таблица колонок, собранная из Документ, который объясняет поля и формат событий источника:
контракта схемы. По ней пишется сторона хранилища — как в бою по документации [`docs/formats/clickstream-event.md`](docs/formats/clickstream-event.md).
источника. Правится только контракт, документ пересобирается. По нему строится прием данных в хранилище. Документ собирается из
_Избегать_: описание схемы, документация контракта описаний колонок генератора: правят их, затем пересобирают документ.
_Избегать_: описание схемы, документация контракта.
**Канонический сериализатор**:
Единственная граница, где запись генератора превращается в байты. Фиксированный
порядок ключей и строк — основа побайтовой воспроизводимости.
**Проигрыватель**:
Компонент доставки готового потока дня в приёмник. Два режима: пакетный
(пачкой, без темпа) и живой день.
**Слепок**:
Полная выгрузка заказов окна изменяемости, снятая бэкендом на границе суток:
состояние заказов на этот момент, а не поток их изменений. Опоздавший заказ
может отсутствовать в ранних слепках; после первого появления ездит до конца
своего окна.
**Дата слепка**:
Дата завершившегося модельного дня, состояние которого снято на исходящей
границе суток. Одна для всех записей выгрузки; не дата заказа и не время
отправки сообщения.
**Время аудита источника**:
Время создания или последнего изменения строки заказа по часам базы источника.
В записи слепка это `created_at` и `updated_at`; бизнес-время покупки живёт в
событии `purchase` отдельно.
**Судьба заказа**:
Исход, моменты изменений, вычеркнутая позиция и опоздание заказа — всё
решается бросками при его рождении и укладывается в окно изменяемости целиком.
Слепок любого дня — чтение готовой судьбы, а не накопление состояния.
**Окно изменяемости**:
Сколько модельных дней заказ ещё может измениться и потому продолжает ездить
в слепках. Константа мира: семь дней. За окном заказ замёрз, выручка дня
перестала «дышать».
**Пакетный забор**:
Способ приёма топика, при котором данные тянет запрос по команде, а не
матвью непрерывно. Так стенд принимает заказы: у слепка есть начало и конец,
и забирать его уместно тогда, когда он собран целиком. Противоположность —
потоковый приём событий.
_Избегать_: путать с приёмником и пакетным режимом — те про сторону
генератора, про отправку; забор — про сторону хранилища, про чтение.
**Техническая проверка**: **Техническая проверка**:
Проверка, чей оракул — внутренний эталон стенда (паспорт мира), которого у Проверка по внутреннему эталону стенда, например сравнение числа событий
настоящего магазина нет. За учебный урок DQ не выдаётся. с паспортом мира. Проверка качества данных (DQ) опирается на данные
_Избегать_: DQ-проверка — та работает на наблюдаемых данных магазина. магазина и связи между ними; ей не нужен эталон генератора, которого
у настоящего магазина нет.
_Избегать_: DQ-проверка в значении проверки по паспорту мира.
### Заказы
**Слепок**:
Дневная выгрузка состояния заказов из серверной части магазина.
Она содержит заказы, оформленные за последние семь модельных дней,
с учетом изменений к концу дня. Опоздавший заказ может отсутствовать
в первых слепках.
**Дата слепка**:
Дата модельного дня, на конец которого снято состояние заказов;
поле `snapshot_date` одинаково во всех строках выгрузки. Даты оформления
заказов могут различаться; выгрузка отправляется на следующий модельный день.
**Время аудита источника**:
Время создания и последнего изменения записи заказа по часам базы
магазина: поля `created_at` и `updated_at`. Время покупки в событии
`purchase` записывает счетчик сайта, и оно может отличаться от этих полей.
**Окно изменяемости**:
Семь модельных дней, включая день оформления, в течение которых заказ
может измениться и входит в слепки. После этого его состояние окончательно.
Например, заказ от 1 июня входит в окно слепков за 1–7 июня; опоздание
сокращает число его появлений, но не сдвигает конец окна.
**Пакетный забор**:
Чтение сообщений Kafka отдельным запросом по команде. Так хранилище
принимает заказы после отправки всего слепка. События сайта оно принимает
непрерывно через материализованное представление.
_Избегать_: смешивать с пакетным режимом генератора: забор — чтение
на стороне хранилища, режим генератора — отправка.
### Модельное время и запуск стенда
**Мир**:
Модель интернет-магазина: люди, их посещения сайта, товары и заказы,
которые создает генератор.
**Ось модельного времени**:
Календарь магазина в генераторе. Первый день, D0, — 1 июня 2026 года,
понедельник (`world.ORIGIN`); следующие дни отсчитываются от него.
Реальное время — часы компьютера, на котором работает стенд. Дата запуска
не меняет даты событий: D0 остается 1 июня, даже если стенд запущен осенью.
**Граница суток**:
Конец одного модельного дня и начало следующего, полночь по часовому
поясу счетчика. Здесь заканчиваются визиты и снимается состояние заказов
для слепка. Даги времени продвигают мир только на целые дни.
**Стартовый мир**:
Данные первых восьми модельных дней, с понедельника по следующий
понедельник, которыми стенд наполняется при первом `make up`.
Команда вызывает `world_initialize`: если мир уже готов и сохранен,
повторный запуск оставляет данные и позицию времени прежними.
`make rebuild-storage` пересоздает стартовый мир. При неизменных коде,
каталоге и настройках его данные повторяются и служат эталоном проверок.
_Избегать_: зерновой мир.
**Паспорт мира**:
Файл для сверки с эталонными данными: `data/world-inventory.json`.
В нем записаны зерно генерации, версия генератора, контрольная сумма
каталога, число событий и контрольная сумма каждого начального дня,
контрольные суммы слепков и счетчики расхождений по заказам.
В git хранится паспорт; сами события и слепки пересчитываются генератором.
_Избегать_: опись мира, манифест, мини-манифест.
**Пошаговый режим**:
Продвижение модельного времени по явной команде: запустить генерацию
следующего дня или нескольких дней.
**Пакетный режим**:
Отправка событий готового модельного дня без пауз между ними.
Используется для начального наполнения, пересборок и проверок.
**Живой день**:
Постепенная отправка событий по времени их появления в модельном дне,
с ускорением: по умолчанию сутки занимают около 24 реальных минут.
Чтобы следующие дни запускались автоматически, снимают с паузы `world_live`.
**Даги модельного времени**:
Наборы связанных задач Airflow для генерации следующих дней магазина:
`world_next_day`, `world_live_day` и `world_live`. Первые два выполняют
шаг по времени, третий автоматически запускает живые дни, пока снят с паузы.
Короткое название — даги времени.
_Избегать_: пульт мира, управляющие даги.
**Шаговый даг**:
Даг, который отправляет данные следующих модельных дней и после успеха
записывает новую позицию времени. `world_next_day` отправляет их пачкой,
`world_live_day` — один день с ускорением. У них нет расписания;
их запускают вручную или из другого дага.
_Избегать_: работник.
### Устройство генератора
**Зерно генерации**:
Число, которое задает повторяемую последовательность случайных выборов
генератора. Одинаковые зерно и день дают одинаковые данные при неизменных
коде, каталоге и настройках. Эталон использует `CANONICAL_SEED` из `seeds.py`.
**Состав мира**:
Люди в модели, их идентификаторы посетителей и заранее заданные привычки:
когда они приходят, как часто возвращаются, склонны ли покупать.
У части людей две куки — это двухкуковые пары.
_Избегать_: состояние мира.
**План состава**:
Расчет того, какие посетители придут в каждый день и какие свойства у них
будут. В `plan.py` из зерна и номера дня получаются новые посетители,
их возвраты и назначенные заказы людей с двумя куками.
**Когорта дня**:
Группа людей, впервые пришедших в один день, со всеми их куками и днями
активности. Генератор рассчитывает ее по зерну и номеру первого дня.
**Приток**:
Появление новых посетителей (`ClientID`) по мере продвижения модельного
времени. Поэтому число разных посетителей за весь период растет и
не совпадает с числом посетителей за отдельный день.
**Хвост возвратов**:
Срок от первого появления человека, в пределах которого его куки могут
возвращаться на сайт. У человека с двумя куками конец срока общий.
**Предыстория**:
Люди, впервые пришедшие до D0, и план их активности. До D0 события
не создаются; возвраты этих людей дают повторных посетителей уже в первый
день генерируемых данных.
**Покупатель**:
Человек, которому генератор заранее назначил повышенную склонность
покупать. Он чаще доходит до заказа, но иногда покупают и остальные.
Из таких людей выбираются те, у кого будут две куки.
_Избегать_: покупатель в значении любого посетителя с покупкой за день;
для этого случая — «визит с заказом».
**Уровень спроса**:
Свойство товара, от которого зависит, как часто после просмотра карточки
его добавляют в корзину. В колонке `demand` каталога три значения:
`магнит`, `обычный`, `залёживается`. Они постоянны по дням и задают
различия в конверсии товаров и брендов.
**Конфигурация мира**:
Числовые настройки модели магазина в `world.py`: приток, частота возвратов,
доли покупателей, начальная дата и другие параметры. Изменение этих
настроек меняет генерируемые данные.
**Подпоток**:
Отдельная последовательность случайных выборов для части генератора,
например для торговли в заданный день. В `seeds.py` она определяется
общим зерном, днем и компонентом; для состава мира есть отдельные
последовательности. Результат не зависит от порядка расчета частей.
**День-функция**:
Расчет событий одних модельных суток по зерну и номеру дня, без генерации
предыдущих дней. Функция `day.stream(seed, day)` возвращает события
в порядке времени.
**Судьба заказа**:
Заранее рассчитанные изменения заказа: итоговый статус, время оплаты
или отмены, удаленная позиция и задержка первого появления в слепке.
Они укладываются в семидневное окно заказа. Слепок показывает состояние
на выбранный день без проигрывания предыдущих изменений.
**Контракт схемы**:
Описания колонок события в модуле генератора `schema.py`: имена,
типы и ограничения. По ним генератор создает и проверяет события,
а `schema_doc.py` собирает описание выгрузки для хранилища.
**Канонический сериализатор**:
Код в `serialize.py`, который превращает события и записи заказов в JSON
для отправки. Порядок ключей и формат значений фиксированы, поэтому
одинаковые записи дают одинаковые байты.
**Проигрыватель**:
Код в `player.py`, который доставляет готовые события дня в приемник,
например в Kafka или файл. В пакетном режиме он отправляет их без пауз,
в режиме живого дня — с ускорением по времени событий.
+24 -32
View File
@@ -1,18 +1,17 @@
"""Трансформации слоя DDS: сборка моделей. """Трансформации DDS: заказы, сессии и карта идентичностей.
Заказ и сессия собираются заменой дневных партиций. Карта идентичностей Запросы рассчитывают заказы и сессии за выбранные дни и записывают результат
пересчитывается целиком после заказа, потому что читает его текущую модель. в промежуточные таблицы. Затем готовые дневные разделы этих таблиц заменяют
разделы основных таблиц. Пока идет расчет, читатели видят прежние данные.
Расписания у дага нет и параметров тоже: объём каждого прогона он выводит сам Карта идентичностей пересчитывается целиком после заказов. Новые строки
из состояния слоёв и позиции мира, поэтому один и тот же запуск одинаково добавляются с меткой времени сборки; представление с FINAL выбирает последнюю
верен из триггера дневного конвейера, с кнопки менти и после падения версию каждой строки.
(docs/architecture/etl.md, «Объём прогона»).
Заказ и сессия гасят повтор заменой партиции целиком: день собирается в Даг запускается без расписания и параметров. Даты он выбирает по состоянию
таблицу-двойник и подменяет собой партицию цели. Карта пишет полный набор пар таблиц и позиции мира. Новый запуск заново определяет диапазон: наличие даты
новой версией, а точное чтение даёт FINAL. Замена партиции разобрана в не подтверждает полноту дня и не гарантирует повтор его сборки после сбоя.
sql/dds/order_replace.sql; точное чтение карты описано в Подробности — docs/architecture/etl.md, «Объём прогона» и «Идемпотентность».
sql/ddl/50-dds-views.sql.
""" """
from __future__ import annotations from __future__ import annotations
@@ -257,16 +256,11 @@ def _days(first_day: datetime.date, last_day: datetime.date) -> list[str]:
@task @task
def replacements(scope: dict[str, object]) -> list[dict[str, object]]: def replacements(scope: dict[str, object]) -> list[dict[str, object]]:
"""Разложить дни в аргументы размноженной задачи замены. """Подготовить параметры отдельной задачи replace для каждого дня.
Шаг существует ради ограничения Airflow: размножать задачу можно expand_kwargs получает список словарей и создает по одному экземпляру
только по целому возврату другой задачи — ни по ключу словаря, ни по задачи на каждый словарь. Дата подставляется в SQL шаблоном Jinja:
карте поверх него. Отсюда отдельный возврат под один список. параметр ClickHouse в ALTER ... PARTITION здесь не поддерживается.
День едет в запрос подстановкой Jinja, а не параметром ClickHouse:
в ALTER … PARTITION сервер параметр не подставляет — замерено,
отвечает синтаксической ошибкой. Значение при этом не внешнее, его
вернул сам ClickHouse запросом объёма.
""" """
return [{"params": {"day": day}} for day in scope["days"]] return [{"params": {"day": day}} for day in scope["days"]]
@@ -276,8 +270,8 @@ def replacements(scope: dict[str, object]) -> list[dict[str, object]]:
schedule=None, schedule=None,
start_date=START_DATE, start_date=START_DATE,
is_paused_upon_creation=False, is_paused_upon_creation=False,
# Два прогона разом собирали бы один и тот же двойник: второй затёр бы # Одновременные запуски используют одни промежуточные таблицы.
# чужую сборку между вставкой и заменой партиций. # Второй мог бы очистить результат первого до замены партиций.
max_active_runs=1, max_active_runs=1,
template_searchpath=str(SQL_ROOT), template_searchpath=str(SQL_ROOT),
tags=["dds"], tags=["dds"],
@@ -298,7 +292,7 @@ def dds_transform():
@task_group(group_id="order") @task_group(group_id="order")
def order(): def order():
"""Заказ: дышащее окно плюс дни, которых в слое ещё нет.""" """Заказ: изменяемые дни последнего слепка и пропущенные дни."""
@task @task
def scope() -> dict[str, object]: def scope() -> dict[str, object]:
@@ -306,8 +300,7 @@ def dds_transform():
query = (SQL_ROOT / "dds" / "order_scope.sql").read_text(encoding="utf-8") query = (SQL_ROOT / "dds" / "order_scope.sql").read_text(encoding="utf-8")
hook = ClickHouseHook(clickhouse_conn_id=CLICKHOUSE_CONNECTION) hook = ClickHouseHook(clickhouse_conn_id=CLICKHOUSE_CONNECTION)
row = hook.get_first(query) row = hook.get_first(query)
# Пустой ответ означает пустой источник: собирать нечего, и # Без заказов ODS расчет невозможен: задача сообщает об ошибке.
# зеленеть на этом нельзя — прогон встаёт с внятной причиной.
if row is None: if row is None:
raise AirflowException( raise AirflowException(
"в ods.order нет ни одного заказа: собирать нечего; " "в ods.order нет ни одного заказа: собирать нечего; "
@@ -342,9 +335,9 @@ def dds_transform():
do_xcom_push=False, do_xcom_push=False,
) )
# Замена идёт по партиции за раз — другой формы у операции нет. День # На каждый день создается отдельная задача замены партиции.
# становится отдельной задачей: в интерфейсе видно, сколько дней несёт # Неудачную замену можно повторить, пока промежуточная таблица
# прогон, а повтор после отказа стоит одного дня, а не всей пересборки. # сохраняет рассчитанные строки этого запуска.
replace = SQLExecuteQueryOperator.partial( replace = SQLExecuteQueryOperator.partial(
task_id="replace", task_id="replace",
conn_id=CLICKHOUSE_CONNECTION, conn_id=CLICKHOUSE_CONNECTION,
@@ -370,9 +363,8 @@ def dds_transform():
"проверьте, дошёл ли поток до хранилища" "проверьте, дошёл ли поток до хранилища"
) )
first_day, last_day = row first_day, last_day = row
# Пропуск, а не отказ: слой полон по последний прожитый день, а # Для каждого прожитого дня уже есть строки сессий.
# сегодняшний ещё идёт. Это обычное состояние повторного прогона, # Новых дней для расчета нет: группа задач будет пропущена.
# и в интерфейсе оно видно пропущенной группой.
if first_day is None: if first_day is None:
raise AirflowSkipException( raise AirflowSkipException(
f"сессии собраны по {last_day} включительно, " f"сессии собраны по {last_day} включительно, "
+12 -8
View File
@@ -1,9 +1,12 @@
"""Трансформации слоя DM: сборка стендовых витрин. """Трансформации слоя DM: сборка стендовых витрин.
Выручка наследует ритм заказов. Сверка соединяет окно заказов с прожитым Даты для расчета выручки берутся из последнего слепка заказов и списка
хвостом событий. Обе заменяют дневные партиции. Трафик наследует пропущенных дней. Сверка покупок с заказами учитывает также прожитые дни
ретроспективность карты идентичностей и пересчитывает всю историю новой событий и дни, где заказ еще не найден. Запросы записывают результаты
версией. Объем каждого прогона выводится из состояния слоев. в промежуточные таблицы, затем заменяют дневные разделы основных таблиц.
Трафик пересчитывается за всю историю: новая связь посетителя с пользователем
магазина меняет число известных пользователей и за прошлые дни.
""" """
from __future__ import annotations from __future__ import annotations
@@ -62,10 +65,11 @@ def _days(first_day: datetime.date, last_day: datetime.date) -> list[str]:
@task @task
def replacements(scope: dict[str, object]) -> list[dict[str, object]]: def replacements(scope: dict[str, object]) -> list[dict[str, object]]:
"""Разложить дни в аргументы размноженной задачи замены. """Подготовить параметры отдельной задачи replace для каждого дня.
Отдельный словарь нужен каждому экземпляру ``expand_kwargs``; тот же expand_kwargs получает список словарей и создает по одному экземпляру
прием разобран в ``dds_transform.py`` на замене партиций заказов. задачи на каждый словарь. Дата подставляется в SQL шаблоном Jinja:
параметр ClickHouse в ALTER ... PARTITION здесь не поддерживается.
""" """
return [{"params": {"day": day}} for day in scope["days"]] return [{"params": {"day": day}} for day in scope["days"]]
@@ -141,7 +145,7 @@ def dm_transform():
@task_group(group_id="purchase_vs_orders") @task_group(group_id="purchase_vs_orders")
def purchase_vs_orders(): def purchase_vs_orders():
"""Сверка: окно заказов, прожитый хвост и предварительные дни.""" """Сверка: изменяемые заказы, прожитые покупки и дни ожидания заказа."""
@task @task
def scope() -> dict[str, object]: def scope() -> dict[str, object]:
+91 -84
View File
@@ -1,111 +1,118 @@
# Модель сессии в DDS # Модель сессии в DDS
Учебный результат: менти строит расчётную сущность, которой нет в потоке, — Сессия объединяет события одного визита. На стенде вы учитесь определять
режет события на сессии по правилу визита, чеканит идентификатор и сверяет ее границы по событиям, вычислять устойчивый идентификатор и сверять
свою нарезку с эталоном. Стенд здесь играет роль сессионизатора: в бою её результат с эталоном. Поэтому сборка читает `dds.event_v`, где нет готового
играет Метрика, которая режет визиты у себя и с октября 2025 отдаёт готовый `VisitID`. Этот идентификатор источника остается в ODS для проверки.
`VisitID` прямо в хитах. Поэтому стенд считает сессии сам, а `VisitID`
генератора использует только как эталон самопроверки — иначе сборка сессий
выродилась бы в `GROUP BY` по чужому полю и центральный урок
слоя пропал бы. Без этого абзаца действия стенда выглядят странно: «зачем
считать то, что уже дано колонкой».
## Зерно ## Зерно
Одна строка — один визит ([CONTEXT.md](../../../CONTEXT.md)): подряд идущие Зерно описывает, чему соответствует одна строка: здесь это один
события одной куки без пауз длиннее 30 минут, не пересекающие границу суток. [визит](../../../CONTEXT.md#люди-и-события). В него входят подряд идущие
В нарезку входят все три типа событий — торговые садятся на свои страницы события одного `client_id` без пауз длиннее 30 минут и без перехода через
и держат сессию наравне с просмотрами. События внутри сессии модельную полночь. Пауза ровно в 30 минут новую сессию не открывает.
упорядочиваются по метке времени события (`UTCEventTime` — опорная колонка Учитываются все три типа событий: `pageview`, `add_to_cart`, `purchase`.
приёма).
Граница суток проходит по модельному дню счётчика, и он у события уже есть: День сессии `session_date` берется из `event_date` событий DDS — это
`EventDate` — свойство содержимого. Раз сессия сутки не пересекает, у всех `EventDate` источника, дата по часовому поясу счетчика. У всех событий
её событий `EventDate` один; он и становится **днём сессии** — колонкой одной сессии она одинакова. Пересчитывать ее из времени UTC не нужно.
партиции, без пересчёта.
SQL рассчитывает границы визитов двумя оконными функциями: первая отмечает
начало нового визита, вторая по этим отметкам присваивает событиям номер
визита. Обе используют порядок `(event_time, event_id)`: при равном времени
события идут по идентификатору. Иначе два расчета могли бы по-разному
упорядочить события и отнести отметку начала к разным строкам.
## Идентификатор ## Идентификатор
Идентификатор сессии в природе не существует: сессия — расчётная сущность, `session_id` вычисляется как `cityHash64(client_id, min(event_time))`.
и id ей чеканит тот, кто сессионизирует. `VisitID` — такой же суррогат, При повторной сборке тех же событий посетитель и начало визита остаются
отчеканенный генератором в роли Метрики; наш — `session_id`, прежними, поэтому идентификатор не меняется. Счетчик или случайный UUID
детерминированный хеш от пары «кука + начало сессии» (`cityHash64`). выдали бы другие значения.
Детерминированность обязательна: партиция дня пересобирается заменой, и
повторный прогон обязан отчеканить те же id — счётчик или UUID это правило
ломают.
`session_id` — не ключ сортировки и не ключ шардирования, а колонка-ручка: По `session_id` можно сопоставить сессии до и после пересборки и назвать
в соединениях он не участвует — события его не несут, мост к заказам живёт конкретную строку при разборе расхождения. Он не используется для сортировки
на событиях. Читатели ручки — лаба сессий (сопоставить свою нарезку со или шардирования. У событий этого поля нет; связь с заказами строится
стендовой и назвать расхождение одним числом) и разбор DQ-расхождений. по событиям покупки.
## Состав строки ## Состав строки
Боевой образец — таблица визитов Метрики, парная к нашему формату хита; по - `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`, начало и конец сессии, В готовом SQL страницы выбираются через `argMin` и `argMax`, а UTM,
длительность, день сессии. Времена — в поясе счётчика по правилу слоя; устройство и город — через `argMin`, только по `event_time`. При одинаковом
- **счётчики**: событий всего, просмотров, корзин, покупок; времени выбор между событиями не определен. Полный порядок окон нарезки
- **страницы**: входная и выходная — как `StartURL`/`EndURL` визита; полного сам по себе эту неоднозначность не устраняет. Это ограничение учитывается
списка страниц нет и у Метрики, топы считаются по событиям; в [задании ступени 4](../../labs/4-1-session-pipeline.md).
- **атрибуция**: UTM-метки первого события сессии — так визит атрибутирует
Метрика. В v1 витрина брала последнее касание — заменено осознанно;
- **паспорт куки**: категория устройства (расшифрованная, язык бизнес-модели)
и город — у куки они постоянны, берутся с первого события;
- **служебные**: `_load_id` и `_load_ts` собравшего запуска — правило слоя
([модель заказа](order.md), «Служебные колонки»).
Осознанные расхождения с образцом: **денег в сессии нет** — визит Метрики В сессии нет суммы выручки: ее считают по заказам магазина. Клиентская
несёт клиентскую выручку, у нас деньги считает только бэкенд (спека, раздел сумма покупки используется в сверке с заказами и в учебной UTM-витрине.
7), а клиентская сумма живёт в сверке; **целей нет** — их место заняли
счётчики торговых событий.
## Движок и наполнение ## Движок и наполнение
`dds.session_rep` — `ReplicatedMergeTree` без колонки версии, по образцу `dds.session_rep` хранит готовые сессии в `ReplicatedMergeTree` без колонки
заказа: партиция — день сессии, ключ сортировки — кука и начало сессии. версии. При чтении не нужно заново выполнять оконный расчет. Партиция —
`dds.session_dist` шардирует по `cityHash64(client_id)`, тем же выражением, раздел таблицы за один `session_date`; ключ сортировки — посетитель
что события, — это ко-локация с `dds.identity_map_rep` ([дока и начало сессии. Читайте результат через `dds.session_v`: представление
хранилища](../storage.md), «Раскладка по шардам»). над `dds.session_dist` явно перечисляет деловые и служебные колонки.
Совпадение ключа с событиями не случайно: все события куки лежат на одном
шарде, поэтому сессионизация локальна по природе — капкан смены ключа при
вставке, который дока хранилища обещает сборкам DDS, у сессий не стреляет.
Сборка — дневной батч: хранилище выше ODS по смыслу батчевое, раз второй Источник сборки — `dds.event_v`. События, сессии и карта идентичностей
источник приходит раз в сутки. Собирается день, который мир уже прожил, и распределены по одному ключу `cityHash64(client_id)`: данные посетителя
собирается один раз и целиком; какой день прожит, слой спрашивает у позиции находятся на одном шарде. Роли объектов и правило записи через `_dist`
мира ([дневной конвейер](../etl.md), «Объём прогона»). Идемпотентность держит описаны в [справочнике хранилища](../storage.md#раскладка-по-шардам).
замена партиции. В отличие от заказов, где батч пересобирает до K «дышащих»
дней, у сессии дышащего окна нет: прожитый день самодостаточен (правило
спеки, раздел 6). Второй раз он соберётся только
вынужденно — когда упавшая замена оставила позади дыру и хвост от неё накрыл
соседние дни. Дней в одном прогоне при этом бывает несколько — столько,
сколько мир успел прожить без слоя. Внутри живого дня
модель сессии про сегодня честно молчит — «сегодня» смотрят через
`dds.event_v`; сессия и по смыслу не финальна, пока день не прожит.
Публичная поверхность — `dds.session_v`: обычное представление с явным Три SQL-файла выполняются последовательно в группе `session` дага
списком деловых и собственных служебных колонок над `dds.session_dist`. `dds_transform`:
Роли объектов и доступ следуют общей [конвенции](../storage.md), «Имена
объектов».
Источник сборки — `dds.event_v` ([контракт события](event.md)): сессии 1. `session_scope.sql` находит первую дату событий, за которую еще нет
нужны от него ключи нарезки, тип события, страница, UTM и паспорт куки. сессий. Последний день для расчета — первый день событий ODS
Механика запуска — где собирается партиция и как подменяется на кластере — плюс `world_position` минус один; `world_position` — номер следующего
не входит: это устройство [дневного конвейера](../etl.md). дня для генерации. Так в расчет попадают только полностью
прожитые дни. При пустом 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` раздел за каждый
выбранный день готовым разделом `dds.session_stage_rep`. Для каждого
дня Airflow создает отдельную задачу `replace`.
Когда все события завершенного дня приняты в ODS, их набор на стенде
больше не меняется. Поэтому готовые сессии обычно не пересчитываются. Если пропущен ранний
день, диапазон включает и все последующие прожитые дни, даже уже собранные.
Если за день сохранилась хотя бы одна сессия, поиск пропущенных дней
не обнаружит потерю остальных. Условия повторного запуска и замены
разделов описаны в [справочнике конвейера](../etl.md#идемпотентность).
Сессий еще идущего дня в модели нет; его события доступны через
`dds.event_v`. Перед выбором дней даг ждет завершения доставки событий
в ODS — см. [готовность источника](../etl.md#готовность-источника).
## Сверка с эталоном ## Сверка с эталоном
Счётная DQ-проверка с зерном «кука × день»: число сессий нашей нарезки Проверка `sessions_vs_reference` в `dq_check` сравнивает для каждого
равно `uniqExact(VisitID)` по тем же событиям. Счёт по Distributed ловит и посетителя и дня число строк `dds.session_v` с `uniqExact(VisitID)`
склеенную, и разрезанную сессию — мимо него проходит лишь взаимная из `ods.event_v`. Эталон читается из ODS, потому что `dds.event_v`
компенсация ошибок внутри одной куки за день; эталон в модель не тащим — скрывает `VisitID`.
строка сессии остаётся нашей, а не генераторной. Поштучное сопоставление
границ — материал лабы, где менти разбирает найденное расхождение руками. Дом проверки — Нулевое число расхождений подтверждает только число сессий каждого
контур DQ этапа 4. Сверке нужен `VisitID`, а сборке сессий — нет; контракт посетителя за день. Например, ошибочное объединение двух визитов
`dds.event_v` его скрывает, и за эталоном проверка идёт в `ods.event_v` и разделение другого на два сохранят их общее число. Поэтому в заданиях
([контракт события](event.md)). вы дополнительно сопоставляете границы сессий по исходным событиям и проверяете страницы,
UTM, устройство и город.
## Читатели ## Читатели
+168 -141
View File
@@ -1,179 +1,202 @@
# Слой DM: витрины # Слой DM: витрины
Как модель DDS превращается в витрины для потребителя BI: какую форму Витрины готовят данные DDS для отчетов и дашбордов. Здесь описаны форма
принимает витрина, кто какие витрины строит и как слой проверяется. Состав хранения, правила обновления и проверки. Состав колонок задан в
колонок каждой витрины задаёт раздел 7 мастер-спеки; здесь — форма, [спецификации](../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)). ([исследование](../research/2026-08-22-view-predicate-pushdown.md)).
Мы все равно оставили материализованные агрегаты ради практики: без них в Готовые таблицы нужны в программе для практики загрузки, обновления
программе не было бы загрузки и пересборки витрин. Представления дают другой и включения витрин в конвейер.
случай: преобразование строк на прежнем зерне.
## Что готово в стенде, а что остаётся заданием ## Что готово в стенде, а что остаётся заданием
Стенд привозит готовыми витрины трёх сюжетов дашборда этапа 8 и Готовы три витрины дашборда и сводка качества:
инфраструктуру контроля:
- `revenue_daily` — выручка по дням и категориям; образец материализованной - `revenue_daily` — одна строка на день заказа и категорию товара;
витрины, на него ссылаются учебные задания; образец дневной пересборки для учебных заданий.
- `purchase_vs_orders` — сверка A+C с классами расхождений; зерно — заказ, - `purchase_vs_orders` — одна строка на заказ: сопоставление покупки
шардирование сохраняет ко-локацию заказов по `cityHash64(order_id)`; из кликстрима с заказом магазина и класс расхождения.
- `daily_traffic` — пара «посетители» / «известные пользователи», обогащение - `daily_traffic` — одна строка на день: число посетителей и известных
[картой идентичностей](dds/identity.md); пользователей с учетом [карты идентичностей](dds/identity.md).
- `dq_summary` — сводка проверок качества (ниже). - `dq_summary` — одна строка на день данных и проверку качества.
Остальные витрины спеки нужно построить в учебных заданиях по готовому ТЗ и В [заданиях ступени 1](../labs/README.md) вы строите `events_enriched_v`,
образцу. Разминка — представление `events_enriched_v` (зерно события), затем `top_pages_daily`, `session_overview`, `dq_errors_daily` и `utm_effectiveness`.
агрегаты `top_pages_daily`, `session_overview` и `dq_errors_daily`, последней — Для каждой физической витрины нужно добавить сборку в `dm_transform`.
самая содержательная `utm_effectiveness` с уроком `declared_*`. В финале Представление `events_enriched_v` отдельной задачи не требует.
каждого задания с агрегатом его задачу нужно добавить в `dm_transform`. До
этого конвейер о витрине не знает, после — пересобирает её вместе со
стендовыми. Представление из разминки добавлять некуда: его не нужно
пересобирать. Тексты заданий лежат в [`docs/labs/`](../labs/README.md).
Дашборд этапа 8 читает только готовые витрины стенда, поэтому его приёмка не Стендовый дашборд читает готовые витрины и работает до выполнения заданий.
зависит от домашних заданий. Отдельный дашборд по учебным витринам можно Дашборд по своим витринам можно собрать в
собрать в [факультативном задании](../labs/optional/superset-dashboard.md), [факультативе](../labs/optional/superset-dashboard.md) после ступени 2.
доступном после ступени 2. Устройство стендового дашборда описано в Устройство стендового дашборда — в [справочнике Superset](superset.md).
[справочнике Superset](superset.md).
## Ритм пересборки ## Ритм пересборки
День витрины пересобирается, пока может измениться его источник, — Дни для пересчета выбирают по тому, как меняется источник:
поэтому ритмов три, по числу источников:
- **от событий и сессий** — день замерзает сразу; пересборка догоняет - События и сессии прожитого дня на стенде больше не меняются. Витрины
хвост несобранных дней заменой дневных партиций; по ним рассчитывают показатели от первого пропущенного дня до последнего
- **от заказов** — день дышит, пока входит в последний слепок; пересборка прожитого и заменяют дневные партиции.
заменяет партиции этого отрезка, дни за ним закрыты по построению; - Заказы могут меняться в течение семи модельных дней. Последний слепок ODS
- **от карты идентичностей** — не замерзает никогда: склейка задает диапазон дат для пересборки; пропущенные дни могут расширить его.
ретроспективна, покупка сегодня делает куку известной во всех её прошлых - Новая связь в карте идентичностей позволяет определить пользователя
днях, а карта пересчитывается целиком. по его прежним визитам: один `client_id` мог встречаться и до покупки,
которая связала его с `user_id`.
Поэтому `daily_traffic`, как и карта, пересчитывается целиком.
Поэтому `daily_traffic` наследует у своего источника и ритм, и механику: `daily_traffic` каждый раз добавляет всю историю новой версией
полный пересчёт всех дней каждый прогон в `ReplacingMergeTree(_load_ts)`, в `ReplacingMergeTree(_load_ts)`. Дневных партиций у нее нет, а текущие
как у самой карты. Дневных партиций у неё нет: партиция — единица замены, строки выбирает представление через `FINAL`.
а эта витрина заменяется только целиком. На стенде такая работа остается
быстрой; витрина читает сессии и карту через их публичные договоры.
Пересчитывать точечно только затронутые дни не из чего: для этого надо
знать, какие связки новые, а полный пересчёт карты этот след затирает —
восстановление следа оставлено лабе повышенной сложности (issue #149).
У `purchase_vs_orders` два источника и одна непрерывная пересборка. Заказная У `purchase_vs_orders` два источника. Начало диапазона — самая ранняя
сторона задает изменяемое окно последним принятым слепком ODS, событийная — из трех дат: первая дата оформления заказа в последнем слепке, первый
последний прожитый день с `purchase`. Левая граница возвращает первую дыру и пропущенный день и первый день с `awaiting_order` — покупкой, для которой
самый ранний день с `awaiting_order`: предварительная строка уже есть в заказ пока не поступил. День ожидания проверяется отдельно: строки за него
витрине, поэтому поиск отсутствующего дня ее не заметит. Правая граница уже есть в витрине, поэтому он не считается пропущенным.
доходит до более позднего из двух хвостов. Так один прогон догоняет пакетный Конец диапазона — более поздняя из последней даты заказов в слепке и последнего
разгон длиннее K и не оставляет старое ожидание замороженным. прожитого дня с покупкой. Один запуск пересобирает весь отрезок,
даже если с предыдущего запуска прошло больше семи модельных дней.
Для учебной сводки брака действует отдельное правило: дни выбираются
по времени загрузки, а прежние итоги сохраняются после удаления
источника по TTL. Оно разобрано в
[задании `dq_errors_daily`](../labs/1-4-dq-errors-daily.md).
## Витрины стенда ## Витрины стенда
Публичные представления перечисляют деловые колонки и координаты запуска Представления `_v` явно перечисляют поля для отчетов, а также `_load_id`
`_load_id`, `_load_ts` явно. Новая колонка физической таблицы сама в договор не и `_load_ts` — идентификатор запуска и время записи. Новая колонка
попадет. физической таблицы появится в представлении только после изменения его SQL.
### `dm.revenue_daily_v` ### `dm.revenue_daily_v`
Зерно: день заказа × категория товара. Витрина читает только оплаченные Одна строка содержит итоги оплаченных заказов за день по категории товара.
заказы из `dds.order_v`, синхронно разворачивает три массива позиций и узнает Запрос читает `dds.order_v` и разворачивает массивы артикулов, количеств
категорию артикула через `dic.products`. Один заказ с несколькими позициями и цен в строки позиций заказа. Артикул, количество и цена с одним индексом
категории считается в `orders` один раз. `units` складывает количество единиц, становятся одной строкой. Категория артикула берется из `dic.products`.
`revenue` складывает цену позиции × количество. Скидка, доставка и итог заказа
между категориями не делятся. `aov` равен `revenue / orders` в том же срезе и
хранит два знака после запятой.
Первый запуск строит всю доступную историю. Следующие берут границы дней из `orders` считает разные заказы: несколько позиций одной категории
последнего слепка ODS: только источник знает, какие даты еще может изменить. в заказе не увеличивают этот счетчик. `units` — сумма количества товаров,
Пропущенный оплаченный день расширяет левую границу, чтобы частичный отказ `revenue` — сумма произведений цены на количество. Скидка и доставка
не оставил дыру после следующего прогона. Агрегат при этом читает DDS; ODS в нее не входят. `aov` равен `revenue / orders` для того же дня и категории
отвечает только на вопрос об объеме. с округлением до двух знаков после запятой.
`dm.revenue_daily_rep` нарезана по `report_date`; день собирается в Первый запуск рассчитывает выручку за всю доступную историю. Следующие
таблице-доноре и заменяется через `REPLACE PARTITION`. Цель и донор шардируются берут диапазон дат оформления заказов из последнего слепка ODS.
по `cityHash64(product_category)`. Кроме дня, у витрины только категория, а Если в DDS есть оплаченные заказы за день, отсутствующий в витрине,
шесть категорий стартового мира близки по объему. Более сложный составной ключ левая граница расширяется до него. Агрегат читает DDS; ODS отвечает
распределил бы эту маленькую таблицу без нового урока. только на вопрос об объеме.
В группе `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_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`.
Повтор замены не добавляет копий строк. Ограничения при частичном отказе
описаны в [конвейере](etl.md#идемпотентность).
Основная и промежуточная таблицы распределяют строки по
`cityHash64(product_category)`. На каждый день приходится не больше шести
категорий стартового мира. Для такой маленькой таблицы достаточно простого
ключа по категории.
### `dm.purchase_vs_orders_v` ### `dm.purchase_vs_orders_v`
Зерно — заказ. Покупки из `dds.event_v` сначала сворачиваются по Одна строка сопоставляет события покупки с заказом магазина. События
`purchase_id[1]`: контракт источника обещает один номер и одну объявленную из `dds.event_v` группируются по номеру заказа `purchase_id[1]`:
сумму, а отдельный счет событий оставляет класс `duplicate_event` по формату источника у покупки один номер и одна заявленная сумма.
наблюдаемым. Заказы приходят из `dds.order_v`. Обе стороны записаны Отдельный счет событий позволяет обнаружить повтор покупки
подзапросами, затем соединены `FULL OUTER GLOBAL JOIN`: голая левая и присвоить класс `duplicate_event`. Заказы берутся из `dds.order_v`.
Distributed-таблица вернула бы каждый непарный заказ по разу с каждого шарда
Полное соединение `FULL OUTER GLOBAL JOIN` сохраняет и покупки без заказа,
и заказы без покупки. Обе стороны записаны подзапросами: при прямом чтении
левой Distributed-таблицы каждый непарный заказ повторился бы по числу шардов
([исследование](../research/2026-08-22-distributed-join.md)). ([исследование](../research/2026-08-22-distributed-join.md)).
Непарная сторона остается `NULL` благодаря `join_use_nulls = 1`. При Если для строки нет пары, поля отсутствующей стороны получают `NULL`
`join_use_nulls = 0` ClickHouse заполнил бы ее нулем или пустой строкой и скрыл благодаря `join_use_nulls = 1`. При `join_use_nulls = 0` они заполнились бы
отсутствие источника. Клиентская `declared_revenue` и серверная `items_total` нулями или пустыми строками, которые сложнее отличить от данных источника.
сравниваются как `Decimal(18, 2)`. Классы проверяются по порядку: Клиентская `declared_revenue` и серверная `items_total` сравниваются
как `Decimal(18, 2)`. Классы проверяются по порядку:
`awaiting_order` для покупки без заказа, затем `cancelled`, `lost_event`, `awaiting_order` для покупки без заказа, затем `cancelled`, `lost_event`,
`duplicate_event`, `amount_delta`, `match`. Поэтому отмененный заказ без `duplicate_event`, `amount_delta`, `match`. Поэтому отмененный заказ без
события остается отмененным, а не потерянным. события остается отмененным, а не потерянным.
Витрина нарезана по `order_day`, цель и донор шардируются по Партиции витрины разделены по `order_day`. Основная и промежуточная таблицы
`cityHash64(order_id)`. Диапазон собирается в доноре одной вставкой и распределяют строки по `cityHash64(order_id)`. Запрос рассчитывает весь
заменяется по дням. Первичная сборка охватывает всю историю обоих источников; выбранный диапазон и одной вставкой записывает строки в промежуточную
следующие прогоны идут по двустороннему ритму, описанному выше. таблицу. Затем ее дневные разделы заменяют разделы основной таблицы.
Первичная сборка охватывает всю историю обоих источников;
следующие прогоны выбирают дни по правилам обоих источников, описанным выше.
### `dm.daily_traffic_v` ### `dm.daily_traffic_v`
В строке один день. `visitors` считает разные `client_id` из В строке один день. `visitors` считает разные `client_id` из
`dds.session_v`; `known_users` считает разные `user_id` после левого соединения `dds.session_v`; `known_users` считает разные `user_id` после левого соединения
с полной текущей `dds.identity_map_v`. Анонимная сессия остается в витрине. с полной текущей `dds.identity_map_v`. Анонимная сессия остается в витрине.
Несколько кук одного человека увеличивают `visitors`, но дают одного Несколько кук, связанных с одним пользователем магазина, учитываются
`known_user`, поэтому `known_users <= visitors`. в `visitors` по отдельности, а в `known_users` дают одного пользователя.
Поэтому `known_users <= visitors`.
Сессии и карта лежат по `client_id`, но публичные `_v` скрывают Сессии и карта распределены по `client_id`. Однако на ClickHouse 26.3
`Distributed`-таблицы от локальной подстановки. На ClickHouse 26.3 чтение через публичные `_v` с `distributed_product_mode = 'local'`
`EXPLAIN PLAN` показал чтение обоих шардов на инициаторе: настройка не дало локального соединения. Поэтому в запросе явно указан
`distributed_product_mode = 'local'` локального соединения не дала. Поэтому `GLOBAL LEFT JOIN`. Он сохраняет сессии без известного пользователя;
витрина использует явный `GLOBAL LEFT JOIN` через публичные договоры, а не в `known_users` учитываются только ненулевые `user_id`.
подменяет их физическими таблицами. Это дороже по сети, зато граница слоев Обновление всей истории новой версией описано в разделе
остается явной. Новая связь куки меняет прошлые дни, поэтому каждый запуск [«Ритм пересборки»](#ритм-пересборки).
пишет полную историю новой версией в `ReplacingMergeTree(_load_ts)`, а
публичное представление выбирает текущие строки через `FINAL`.
## `dm.dq_summary` ## `dm.dq_summary`
Зерно публичного `dm.dq_summary_v` — «день данных × атомарная проверка». Одна строка `dm.dq_summary_v` — день данных и одна проверка.
Строка отвечает, каково качество дня сейчас; историю запусков хранит Airflow. Строка отвечает, каково качество дня сейчас; историю запусков хранит Airflow.
Партиция совпадает с днём данных и заменяется через `REPLACE PARTITION`. Партиция совпадает с днём данных и заменяется через `REPLACE PARTITION`.
Деловые колонки сводки: Деловые колонки сводки:
- `data_date` — проверяемый день; - `data_date` — проверяемый день;
- `check_name` — устойчивое имя инварианта; - `check_name` — имя проверяемого правила;
- `reference_rows` — число деловых ключей на эталонной стороне; - `reference_rows` — число деловых ключей на эталонной стороне;
- `actual_rows` — число деловых ключей в проверяемом объекте; - `actual_rows` — число деловых ключей в проверяемом объекте;
- `failed_rows` — число отсутствующих или различающихся ключей; - `failed_rows` — число отсутствующих или различающихся ключей;
- `status` — `pass` при нуле нарушений, иначе `fail`. - `status` — `pass` при нуле нарушений, иначе `fail`.
Служебные `_load_id` и `_load_ts` называют собравший запуск. Представление В `_load_id` и `_load_ts` записаны идентификатор запуска и время расчета
перечисляет колонки явно. Порогов и промежуточного состояния нет. сводки. Даже одно расхождение дает `fail`.
## Звено DQ конвейера ## Звено DQ конвейера
Даг `dq_check` стоит в `etl_pipeline` после `dm_transform` и проверяет пять Даг `dq_check` стоит в `etl_pipeline` после `dm_transform` и проверяет пять
атомарных договоров: правил:
- `sessions_vs_reference`: ключ — `client_id` внутри дня, эталон — - `sessions_vs_reference`: ключ — `client_id` внутри дня, эталон —
`uniqExact(VisitID)` из `ods.event_v`, объект — число строк `uniqExact(VisitID)` из `ods.event_v`, объект — число строк
@@ -192,43 +215,47 @@ Distributed-таблица вернула бы каждый непарный з
сравниваются с прямыми агрегатами `dds.session_v` и сравниваются с прямыми агрегатами `dds.session_v` и
`dds.identity_map_v`. `dds.identity_map_v`.
Область каждой проверки выводится из объединения сторон. Полностью потерянный Каждая проверка берет дни и ключи из обоих сравниваемых наборов данных.
день или ключ поэтому не исчезает из результата. Разные ошибки тоже не могут Поэтому потерянная строка остается видна на другой стороне сравнения.
скомпенсировать друг друга общим итогом: `failed_rows` считается после Расхождения считаются по ключам: завышение одного показателя не скроет
сопоставления деловых ключей. занижение другого в общей сумме за день.
`classes_vs_inventory` — техническая растяжка пути от генератора до DM. Она `classes_vs_inventory` сравнивает результат с паспортом мира. Это техническая
подтверждает, что до витрины доехал записанный в паспорте мир, но не проверяет проверка доставки и обработки эталонных данных стенда. Она не проверяет
сам генератор: у настоящего магазина такого оракула нет. сам генератор; у настоящего магазина такого эталона нет.
Четыре SQL-сверки считают дневные итоги в ClickHouse и возвращают Airflow только Четыре SQL-сверки считают дневные итоги в ClickHouse и возвращают Airflow только
сводку и малый срез диагностики. Техническая сверка с паспортом читает небольшой сводку и несколько примеров расхождений. Техническая сверка с паспортом читает небольшой
файл JSON в Python. Все пять сборок пишут в общий донор. После их успеха даг файл JSON в Python. Результаты всех пяти проверок записываются в общую
заменяет проверенные дневные партиции. Затем именованные задачи читают промежуточную таблицу. Когда расчеты завершились, ее дневные разделы
опубликованную сводку и падают при `failed_rows > 0`. Расхождение успевает заменяют разделы `dm.dq_summary_rep`. Затем отдельные задачи читают
сохраниться до красного завершения, а ошибка сборки не затирает прежний сохраненную сводку и завершаются ошибкой при `failed_rows > 0`.
результат. Красная задача печатает не больше двадцати нарушенных ключей за Так расхождение сохраняется в таблице до отказа задачи. Если сам расчет
проверку со значениями сторон. Для полного списка итоговую агрегацию SQL-сверки не выполнен, прежняя сводка остается на месте. Задача с расхождениями
выводит в журнал не больше двадцати ключей со значениями обеих сторон. Для полного списка итоговую агрегацию SQL-сверки
заменяют на `SELECT * FROM comparison WHERE failed ORDER BY data_date, заменяют на `SELECT * FROM comparison WHERE failed ORDER BY data_date,
business_key`. business_key`.
Если после починки день исчез с обеих сторон, его больше нет и в новом доноре. Если день исчез из обоих сравниваемых наборов, новый расчет не создаст
Такой день удаляется из сводки отдельно: старая красная строка не должна для него строк. Даг отдельно удаляет прежний результат за этот день
превращаться в историю попыток. из сводки.
## Даг самопроверки ## Даг самопроверки
Домашние задания проверяет `homework_check` — отдельный даг без расписания, `homework_check` запускается вручную и содержит задачу для каждого из
который запускают вручную. В нём есть задача для каждого из [пяти заданий по витринам](../labs/README.md). Если витрины нет или значения
[пяти заданий по витринам](../labs/README.md). расходятся с источником, задача завершается ошибкой. Журнал показывает
Задача красная, пока витрины нет или счёт не сходится, и зелёная, когда причину и примеры несовпадений. Успех означает, что проверяемые значения
построено верно; сообщение об ошибке говорит, что не так и куда смотреть. Проверки — инварианты против источников витрины, совпали; остальные требования задания вы проверяете сами.
золотые константы не прибиваются. В `dq_summary` даг не пишет: он инструмент
лабы, а не история качества конвейера.
Приём проверен на соседнем учебном стенде Airflow + Greenplum: Ожидаемые значения рассчитываются по исходным данным текущего стенда.
валидационный даг с группами проверок по слоям давал обратную связь по каждому Для четырех заданий это события или сессии DDS, для сводки брака —
шагу задания, не трогая рабочий конвейер. таблицы ошибок ODS. Сравнение по деловому ключу учитывает отсутствующие
строки с обеих сторон. Результаты не пишутся в `dq_summary`:
самопроверка учебных витрин отделена от качества готового конвейера.
Готовые SQL-проверки открывайте после собственного решения: они содержат
расчет существенной части задания.
## Отклонённые варианты ## Отклонённые варианты
+107 -82
View File
@@ -1,28 +1,23 @@
# Дневной конвейер трансформаций # Дневной конвейер трансформаций
Как принятые источники превращаются в модель DDS и витрины DM: кто запускает Даг `etl_pipeline` собирает модель DDS, обновляет витрины DM и проверяет
трансформации, когда и в каком объёме. Правила самих сущностей — зерно, результат. Здесь описаны запуск, выбор дней и запись результата в таблицы.
движки, что именно пересобирается — живут в доках слоя Состав моделей — в [справочниках DDS](dds/README.md) и [DM](dm.md).
([карта модели DDS](dds/README.md)); здесь — только конвейер.
Дневная сборка вперёд появляется на этапе 4 и остаётся основным режимом
конвейера. На этапе 5 к ней добавляется партиционная переобработка: запуск
произвольного прошлого дня по команде.
## Форма ## Форма
Дневной конвейер на стенде один — даг `etl_pipeline`. Своей работы `etl_pipeline` последовательно запускает `dds_transform`, `dm_transform`
у него нет: он дёргает слойные даги последовательно и ждёт каждый — и `dq_check`. Каждый следующий даг начинается после успеха предыдущего.
`dds_transform`, затем `dm_transform` и `dq_check`. Состав проверок описан в Для этого `TriggerDagRunOperator` ждет завершения запущенного им дага.
[доке слоя DM](dm.md). Если тот завершился ошибкой, задача ожидания в конвейере тоже завершится
ошибкой.
Поведение проверено через Context7 по документации Airflow 3.x (3.1.6).
Ждущие вызовы сделаны `TriggerDagRunOperator`. Проверка через Context7 по У этих дагов нет расписания. Их вызывает конвейер; для отдельного разбора
ближайшей доступной документации Airflow 3.x (версия 3.1.6) подтвердила, что можно запустить даг слоя вручную. У каждого разрешен один активный запуск:
оператор опрашивает точный созданный им прогон и краснеет вслед за потомком. два запуска одного дага используют одни промежуточные таблицы и могли бы
стереть результаты друг друга. Расчет каждой модели выполняется отдельной
Слойные даги устроены по образцу `orders_ingest`: без расписания, один задачей или группой задач.
активный прогон, запускаются триггером конвейера или рукой при разборе
происшествий. Витрины внутри `dm_transform` — отдельные задачи.
## Три пути запуска ## Три пути запуска
@@ -34,83 +29,113 @@ world_live_day ───┼─> etl_pipeline ─> dds_transform ─> dm_transfor
world_next_day ─────> источники без ETL world_next_day ─────> источники без ETL
``` ```
- **Живой день.** `world_live_day` дёргает конвейер с ожиданием — последним - `world_live_day` запускает конвейер после успешной генерации, приема
шагом, после установки позиции. Порядок несёт смысл: день сыгран — факт, заказов и обновления `world_position`. Если трансформация упала, день
упавшая трансформация его не отменяет, поэтому позиция ставится раньше; но уже прожит и позиция не откатывается. После исправления причины
зеленеть, не узнав, разложился ли день, шаговый даг не должен ([ADR запустите `etl_pipeline` вручную.
0003](../adr/0003-dag-run-state.md)). Такой отказ чинится перезапуском - `world_next_day` отправляет дни пачкой и ждет приема заказов, но конвейер
`etl_pipeline` рукой — перезапуску ничего не нужно помнить, см. «Объём не запускает. После него запустите `etl_pipeline` сами.
прогона». - `world_initialize` запускает конвейер последним шагом: при успешном
- **Разгон пачкой.** `world_next_day` конвейер не дёргает: даги модельного первом `make up` начальные данные уже собраны в DDS и DM.
времени управляют генератором и только им. После разгона менти запускает
конвейер сам — и видит, как слои догоняют уехавший вперёд мир.
- **Рождение мира.** `world_initialize` дёргает конвейер последним шагом:
`make up` отдаёт стенд с собранными слоями, а не пустые витрины. Уроки
ручного запуска начинаются со следующих дней.
## Объём прогона ## Объём прогона
Параметров у конвейера нет: объём он выводит сам из `world_position` и Объем прогона — дни, которые обрабатывает один запуск. Даты передавать
состояния слоёв. Сессии догоняют хвост — от первого несобранного дня по не нужно: даги определяют их по состоянию таблиц и `world_position`.
последний прожитый; заказы пересобирают дышащее окно и тот же хвост; карта Например, после генерации семи дней без ETL один запуск рассчитает сессии
идентичностей пересчитывается целиком; правила витрин — у за все семь дней.
[слоя DM](dm.md).
Поэтому один и тот же запуск одинаково верен из триггера, с кнопки менти и
после падения, а разгону на семь дней хватает одного прогона.
Окно заказов задаёт последний слепок, наблюдаемый в ODS. У каждого заказа Для расчета сессий выбираются дни от первого, за который в DDS еще нет
`snapshot_date` равен последнему слепку, в котором источник его показал. сессий, до последнего прожитого.
Трансформация выбирает строки с максимальной `snapshot_date` и берёт среди них `world_position` хранит номер следующего дня для генерации, поэтому
минимальный и максимальный `order_date`. Это границы дней, состояние которых последний прожитый день равен `min(EventDate) + world_position - 1`.
источник ещё может изменить Начало календаря берется из ODS; события ранних дней там не удаляются по TTL.
([ADR 0014](../adr/0014-order-version-snapshot-date.md)). Кроме живого окна, Один `max(EventDate)` не годится: события за день есть и во время его
трансформация закрывает пропуски: находит самый ранний день, представленный в генерации, когда последние визиты еще не закончились.
ODS и ещё не собранный в DDS, затем при необходимости расширяет до него левую Точный запрос — [`session_scope.sql`](../../sql/dds/session_scope.sql).
границу.
Позиция мира заказам не отвечает: она говорит, какой день прожит, а не какой Для заказов диапазон задает последний принятый слепок ODS. Среди строк
доехал. Прогон дня D везёт слепок дня D − 1, да и ставится позиция раньше с максимальной `snapshot_date` выбираются самая ранняя и самая поздняя
трансформации, так что после неудачного приёма обгоняет хранилище. Слой обязан даты оформления заказа. Состояние заказов за эти дни еще может измениться.
считать объём по тому, что у него есть, а не по тому, что мир о себе заявляет. Если в ODS есть заказы за более ранний день, а в DDS за него нет строк,
начало диапазона сдвигается до этого дня. Запрос заново рассчитывает
заказы за все дни выбранного отрезка.
Для сессий границу задаёт `world_position`: она отличает полностью прожитый Позиция мира для выбора дней заказов не подходит: с событиями дня D
день от ещё идущего. Сами события этой границы не показывают: и посреди дня D, приходит слепок за D−1. Объем рассчитывается по принятым строкам ODS;
и сразу после его конца их позднейший день равен D. Если собрать сессии посреди их наличие еще не доказывает полноту слепка.
дня, визит, начатый в 23:50, останется обрезанным: завершённый день слой
повторно не собирает. Поэтому последний прожитый день равен первому дню Карта идентичностей пересчитывается целиком. Витрины выбирают дни
событий плюс позиция минус один по своим источникам — см. [ритм пересборки DM](dm.md#ритм-пересборки).
([`sql/dds/session_scope.sql`](../../sql/dds/session_scope.sql)). Если за день сохранилась хотя бы одна строка, запрос не считает его
Оговорка про «доехал» остаётся в силе и здесь: за хвост потока отвечает не пропущенным. Полноту данных проверяют отдельно.
позиция, а «Готовность источника».
## Готовность источника ## Готовность источника
Первый шаг `dds_transform` проходит два барьера за общий срок в 300 секунд. Перед сборкой DDS задача `wait_for_source` последовательно ждет двух
Сначала клиент `confluent-kafka` сверяет подтвержденные смещения группы условий:
`clickstream_hits` с верхними смещениями обоих разделов `hits`. Группа, оба
раздела и все смещения должны существовать. Суммарное отставание должно стать
нулевым. Каждый опрос пишет в журнал оба смещения и отставание по разделу.
Только после этого шаг проверяет обе ноды ClickHouse через 1. Группа `clickstream_hits` дочитала оба раздела топика `hits`.
`clusterAllReplicas`: у `ods.event_dist` в `system.distribution_queue` не Подтвержденные смещения сравниваются с верхними смещениями Kafka.
должно остаться файлов. Каждый опрос пишет в журнал ноды, число файлов и Группа и смещения должны существовать, суммарное отставание — равняться нулю.
`last_exception` очереди. Оба барьера опрашиваются раз в три секунды и делят 2. В `system.distribution_queue` у `ods.event_dist` не осталось файлов
один крайний срок. После Kafka срок не отсчитывается заново. доставки на шарды. Запрос через `clusterAllReplicas` читает обе ноды.
Зелёный приём заказов о событиях не говорит ничего — топики разные и Опросы идут с паузой в три секунды; на обе стадии задан общий срок
потребители разные. Одного нулевого отставания Kafka тоже мало: чтец мог 300 секунд. Журнал показывает смещения и отставание Kafka, затем число
подтвердить сообщение раньше, чем Distributed-таблица доставила файл на шард. файлов и `last_exception` очереди на каждой ноде. Если готовность не
Два барьера закрывают обе части пути до того, как сессии выберут свой объем. подтверждена за срок ожидания, задача завершается ошибкой.
Одного нулевого отставания Kafka мало: сообщение может быть подтверждено,
пока Distributed-таблица еще доставляет его на шард. Успех `orders_ingest`
тоже не заменяет эти проверки: у заказов другой топик и потребитель.
Эти условия проверяют доставку на момент опроса. Они не останавливают
поступление новых событий и не сверяют содержимое таблиц. Поэтому
`world_position` отдельно ограничивает сборку сессий прожитыми днями,
а качество результата проверяет `dq_check`.
## Идемпотентность ## Идемпотентность
Партиция не заменяется простой вставкой. День собирается в таблицу-двойник — Идемпотентность означает, что повторная обработка тех же исходных данных
вставка через Distributed раскладывает строки по шардам, — затем не добавляет копии деловых строк. Служебные `_load_id` и `_load_ts`
`REPLACE PARTITION` на кластере атомарно подменяет кусок каждого шарда. Эта при новой сборке меняются.
механика — часть самих трансформаций и строится этапом 4: без неё дневной
прогон не повторяем. Сущности, пересчитываемые целиком, идемпотентны и без Результат дневного расчета сначала записывается в промежуточную таблицу.
замены: полная перезапись в `ReplacingMergeTree` — так живут карта Она нужна, чтобы читатели продолжали видеть прежние данные, пока новые
идентичностей и витрина `daily_traffic` ([слой DM](dm.md)). еще рассчитываются. Например, для сессий это `dds.session_stage_rep`,
а основная таблица — `dds.session_rep`.
Задачи выполняются в таком порядке:
1. `scope` выбирает даты для расчета.
2. `rebuild` очищает промежуточную таблицу, рассчитывает строки за выбранные
дни и записывает их через `_stage_dist`. Настройка
`distributed_foreground_insert = 1` заставляет вставку дождаться
доставки строк на шарды.
3. `replace` заменяет в основной таблице раздел за каждый выбранный день
готовым разделом промежуточной таблицы. Раздел таблицы называется
партицией; для замены используется `REPLACE PARTITION`.
Структура и распределение строк по шардам у двух таблиц должны совпадать.
Замена выполняется на локальных `_rep` с `ON CLUSTER`. На каждом шарде
читатель видит либо старый, либо новый раздел целиком — это атомарность
замены. Общей транзакции для двух шардов и всех дней нет: при сбое один
шард может уже содержать новый результат, а другой — прежний. Повторить
`replace` можно, пока промежуточная таблица хранит тот же результат;
копии строк в основной таблице от этого не появятся.
Новый запуск всего конвейера заново выбирает дни. Он не обязан повторить
все дни неудачного запуска: если дата уже есть в целевой таблице, поиск
пропусков ее не выберет. В частности, изменение SQL сессий само по себе
не заставляет пересчитать заполненные дни. Повторная сборка выбранного
дня разобрана в [задании ступени 4](../labs/4-1-session-pipeline.md).
Карта идентичностей и `daily_traffic` обновляются иначе: каждый запуск
добавляет полный результат новой версией в `ReplacingMergeTree(_load_ts)`.
Публичные представления читают текущие строки через `FINAL`. Старые версии
могут физически оставаться до фонового слияния; повторяемость относится
к результату чтения, а не к числу строк на диске.
## Чего конвейер о себе не знает ## Чего конвейер о себе не знает
+72 -91
View File
@@ -30,115 +30,96 @@ keeper, Kafka, каркас сервисов. Этап 2 собран: в `sql/d
## Имена объектов ## Имена объектов
Имя объекта заканчивается тем, что это за объект: Суффикс показывает назначение объекта:
| Суффикс | Что это | | Суффикс | Что это |
|---|---| |---|---|
| `_rep` | локальная таблица шарда, движок семейства `Replicated*` | | `_rep` | локальная таблица: хранит данные своего шарда, движок семейства `Replicated*` |
| `_dist` | `Distributed` поверх одноимённой локальной | | `_dist` | таблица `Distributed`: читает данные шардов и распределяет запись между ними |
| `_kafka` | таблица на движке `Kafka` | | `_kafka` | таблица на движке `Kafka`: читает сообщения топика |
| `_file` | таблица на движке `File` — чтец файла | | `_file` | таблица на движке `File`: читает файл |
| `_mv` | материализованное представление | | `_mv` | материализованное представление: обрабатывает вставляемые данные |
| `_v` | обычное представление, публичная поверхность чтения | | `_v` | обычное представление: выполняет сохраненный запрос при чтении |
Суффикс носит каждый физический объект. Голого имени у таблицы не существует: Для запросов к моделям DDS и витринам DM используйте `_v`. Представления
запрос к `stg.hits_raw` даёт громкую ошибку «нет такой таблицы» — а под голым явно перечисляют колонки, поэтому добавление поля в физическую таблицу
именем в документах и разговоре понимается сущность, у которой этих объектов не меняет набор полей для читателя. У `dds.event_v` нет отдельной таблицы
несколько. Единственное исключение — словари: у них воплощение одно, шардировать DDS: представление читает события ODS.
нечего, и суффикс ничего не различал бы. В зоне `dic` обе формы стоят рядом и
правило видно целиком: `dic.products_file` — чтец файла, `dic.products` — сам
словарь.
Распространённая конвенция, где голое имя означает локальную таблицу, а В документации имя без суффикса, например `stg.hits_raw`, обозначает
распределённая получает суффикс `_all`, ошибается иначе: забытый суффикс тихо сущность с несколькими объектами. Таблицы с таким именем нет: в запросе
возвращает данные одного шарда. Правило спеки «проверки и контрольные суммы — нужно указать суффикс. Исключение — словари ClickHouse: `dic.products`
только по `Distributed`» такую тишину переживает плохо. обращается к каталогу товаров, а `dic.products_file` читает его файл.
Второе свойство — в списке по алфавиту объекты группируются по сущности, а не по В STG имя берется из топика: `hits_raw`, `orders_raw`. В типизированных
технологии: все четыре объекта топика `hits` стоят рядом, потому что различаются слоях оно называет сущность в единственном числе: `event`, `order`,
хвостом, а не началом имени. Речь про дерево в клиенте и про `ORDER BY name`: `session`. Объекты одной сущности различаются суффиксом и стоят рядом
порядок выдачи `SHOW TABLES` документация не оговаривает. при сортировке имен по алфавиту.
Начало имени — сущность, и берётся она в разных слоях из разных мест. В STG имя `_rep` и `_dist` нужны для загрузки, диагностики и обслуживания. Роль
приходит от транспорта: слой хранит то, что доехало по топику, и зовётся именем `bi_reader` может читать их в ODS, DDS и DM, но для обычного анализа
топика — `hits_raw` от `hits`, `orders_raw` от `orders`. В типизированных слоях принято использовать `_v`. Это соглашение о чтении, а не ограничение прав;
имя приходит от предметной области и стоит в единственном числе: `event`, причина широкого доступа описана в
`order`, `session`. Граница между «как привезли» и «что это такое» [ADR 0007](../adr/0007-clickhouse-access.md) и
проходит по STG, и имена её показывают.
В DDS публичный договор сущности заканчивается на `_v`. Потребительские запросы
и витрины называют только представление; `_rep` и `_dist` служат загрузке,
диагностике и обслуживанию. Представление перечисляет колонки явно, даже если
его запрос — простая проекция над `_dist`: договор не выдаёт устройство
кластера и случайно добавленные в физическую таблицу колонки. У события
физической копии в DDS нет, но правило то же: `dds.event_v` — его публичный
договор.
Право доступа здесь намеренно шире договора. Роль `bi_reader` может читать
любые объекты ODS, DDS и DM, но физические объекты от этого не становятся
публичной поверхностью. Конвенцию держат явный список колонок, документация и
ревью, а не запрет на физические таблицы.
Широкое право принято в [ADR 0007](../adr/0007-clickhouse-access.md): узкая
граница поверх `Distributed` потребовала бы отдельной механики. Механизм
ClickHouse и отклонённые варианты записаны в
[исследовании](../research/2026-08-25-clickhouse-distributed-view-security.md). [исследовании](../research/2026-08-25-clickhouse-distributed-view-security.md).
## Раскладка по шардам ## Раскладка по шардам
Пишем только в `_dist`. Локальные таблицы остаются для чтения и обслуживания — Шард хранит часть строк. При записи через `_dist` место строки выбирает
операций с партициями, ручной переобработки. Правило не про удобство: при записи ключ шардирования. При прямой записи в `_rep` она остается на той ноде,
через распределённую таблицу раскладку определяет ключ шардирования, то есть где выполнена вставка. Поэтому данные записывают через `_dist`,
свойство данных, а при записи в локальную — то, какая нода случайно выполняла а операции с партициями выполняют на локальных `_rep`.
код. Отсюда урок стенда: какая нода читала топик, меняется между прогонами
(видно в колонке `consumer_host`), а куда легли данные — нет.
Ключи шардирования: `cityHash64(ClientID)` у событий, `cityHash64(order_id)` у У событий ключ — `cityHash64(ClientID)`, у заказов —
заказов, `cityHash64` сырой строки у STG и у таблицы ошибок. У первых двух хеш `cityHash64(order_id)`, у STG и таблиц ошибок — хеш сырой строки.
выбран против перекоса: структурированный числовой идентификатор распределяется Хеш распределяет числовые идентификаторы по шардам и направляет одинаковые
по остатку от деления неравномерно. У сырого слоя выбора нет — строку иначе не сообщения на один шард. У сырой строки и разобранного события разные
разложишь; там хеш даёт другое свойство, одинаковые сообщения ложатся на один ключи: они могут оказаться на разных шардах. Счетчики STG и ODS сравнивают
шард. по всему кластеру через `_dist`, а не отдельно по каждой ноде.
Ключи у слоёв разные, и это имеет наблюдаемое следствие: сырая строка и При `INSERT INTO цель_dist SELECT … FROM источник_dist` важна настройка
разобранное из неё событие почти всегда оказываются на разных шардах. `parallel_distributed_insert_select`. При значении `2` и одном кластере
Пошардовые счётчики STG и ODS поэтому не сходятся и сходиться не должны — вставка может выполняться локально на шардах, сохраняя раскладку источника
сверять слои можно только через `_dist`. вместо ключа цели. Если ключи различаются, укажите
`parallel_distributed_insert_select = 0`: нода, принявшая запрос,
соберет результат и распределит его по ключу цели. Так сделано в
[`sql/ods/order_parse.sql`](../../sql/ods/order_parse.sql) и при сборке
выручки. Неверную раскладку проверяют по `hostName()` и ожидаемому шарду
строки; общий счет по `_dist` ее не показывает.
Разные ключи у слоёв ставят капкан на вставке `INSERT INTO цель_dist SELECT … События, `dds.session_dist` и `dds.identity_map_dist` распределены
FROM источник_dist`. Заводское значение `parallel_distributed_insert_select` — по `cityHash64(client_id)` (в ODS поле называется `ClientID`). Данные
2: если обе таблицы на одном кластере, вставка исполняется локально на каждом одного посетителя лежат вместе — это ко-локация. Заказы DDS и их сверка
шарде, минуя ключ шардирования цели. Строки наследуют раскладку источника, и сохраняют раскладку по `cityHash64(order_id)`. Обычное соединение допускается
ошибки при этом нет — расхождение видно только пересчётом ключей по по ключу ко-локации; для остальных соединений применяется `GLOBAL`.
`hostName()`. Так половина заказов и легла на оба шарда, где Само совпадение ключей еще не гарантирует локального выполнения запроса
`ReplacingMergeTree` через границу шарда дубли не схлопывает (замер — «Что через представления — пример есть в [описании `daily_traffic`](dm.md#dmdaily_traffic_v).
проверено», #137). Лекарство — явное `parallel_distributed_insert_select = 0`
в `SETTINGS` такой вставки: раскладку делает инициатор, по ключу цели
(`sql/ods/order_parse.sql`). Капкан ждёт и сборки DDS: они тоже будут читать
`_dist` и писать в `_dist` с другим ключом.
Ключи ко-локации названы заранее, потому что на них стоит политика соединений из Для дневной пересборки DDS и DM результат сначала записывают
раздела 6 спеки: обычное соединение разрешено только по ключу ко-локации, всё в промежуточную таблицу. Основная таблица в это время остается доступна
прочее — через `GLOBAL`. Значит `dds.session_dist` и `dds.identity_map_dist` для чтения с прежними данными. Например, заказы записываются через
шардируются по `cityHash64(client_id)`. Объекты DDS с зерном заказа сохраняют `dds.order_stage_dist` в `dds.order_stage_rep`, затем ее дневные разделы
ко-локацию по `cityHash64(order_id)`; ключи остальных частей модели и витрин заменяют разделы основной `dds.order_rep`.
появляются вместе с ними.
Замена партиций не используется для `ods.order`: версии прошлых дней У промежуточной и основной таблиц должны совпадать структура, ключи
доливаются, а прямое чтение Kafka не задаёт границы полного слепка ([ADR партиции и сортировки, первичный ключ и политика хранения. Поэтому
0010](../adr/0010-order-versions-in-ods.md)). Если партиционная пересборка `dds.order_stage_rep` объявлена через `AS dds.order_rep` с совпадающими
понадобится будущим объектам DDS или DM, партиция-донор должна быть заранее ключами. Требования сверены через Context7 с
разложена по шардам по тому же ключу. Форму донора следует решать вместе с таким [документацией ClickHouse](https://clickhouse.com/docs/sql-reference/statements/alter/partition#replace-partition)
объектом, а не переносить на ODS заранее. для схемы на ClickHouse 26.3. DDL показан в
Первым такой донор понадобился заказу DDS, и правило на нём видно целиком:
`dds.order_stage_rep` объявлен через `AS dds.order_rep` и шардируется тем же
`cityHash64(order_id)`. Совпадение структуры требует сама операция, а совпадение
ключа — устройство замены: она идёт по локальным таблицам, шард за шардом, и
строка, попавшая в доноре не на свой шард, при подмене пропала бы с одного шарда
и задвоилась на другом. Доводы целиком — в
[`sql/ddl/45-dds-tables.sql`](../../sql/ddl/45-dds-tables.sql). [`sql/ddl/45-dds-tables.sql`](../../sql/ddl/45-dds-tables.sql).
Промежуточная таблица должна также распределять строки по тому же ключу,
здесь — `cityHash64(order_id)`. `REPLACE PARTITION` копирует раздел между
локальными таблицами каждого шарда и не перераспределяет строки.
Поэтому уже в промежуточной таблице заказ должен находиться на нужном шарде.
Перед заменой вставка с `distributed_foreground_insert = 1` дожидается
доставки данных на шарды.
В `ods.order` этот способ не применяется: туда добавляют версии заказов
из слепков, а актуальные строки выбираются при чтении
([ADR 0010](../adr/0010-order-versions-in-ods.md)).
## Служебные колонки ## Служебные колонки
Собственные колонки не повторяют имён виртуальных. Виртуальные даёт движок: Собственные колонки не повторяют имён виртуальных. Виртуальные даёт движок:
+24 -42
View File
@@ -1,35 +1,20 @@
-- DDS: нарезка событий на визиты и сборка дней в таблицу-двойник. -- DDS: расчет сессий за дни от first_day до last_day включительно.
-- Сначала очищаем промежуточную таблицу session_stage_rep, затем записываем
-- в нее рассчитанные строки через session_stage_dist. Основная session_rep
-- остается прежней до выполнения session_replace.sql.
-- --
-- Двойник очищается целиком в начале, дни собираются одной вставкой на весь -- Первое окно помечает событие единицей, если с предыдущего события того же
-- отрезок — по правилу заказа (sql/dds/order_rebuild.sql). -- посетителя за тот же день прошло больше 1800 секунд. Второе окно суммирует
-- эти единицы: получается номер визита для каждого посетителя за день.
-- PARTITION BY (client_id, event_date) считает каждый модельный день отдельно:
-- события разных дней не попадут в одну сессию даже при короткой паузе.
-- Учитываются все типы событий.
-- --
-- Нарезка — центральный расчёт слоя, и стоит она на двух окнах подряд. -- У первого события lagInFrame возвращает его собственное время: разница
-- Первое спрашивает у каждого события, сколько прошло с предыдущего события -- равна нулю, opens_visit = 0, и первый visit_no тоже равен нулю.
-- той же куки в тот же день, и ставит единицу там, где пауза длиннее -- Оба окна используют (event_time, event_id), чтобы при равном времени
-- таймаута: это шов между визитами. Второе считает нарастающий итог этих -- отметка начала и номер визита относились к одним событиям.
-- единиц — номер визита внутри дня куки. Дальше визит становится обычной -- Порог 1800 секунд — правило визита; он не передается параметром.
-- группировкой.
--
-- Границу суток окна не проверяют: она уже в разбиении PARTITION BY
-- (client_id, event_date), поэтому визит через полночь не склеится и при
-- короткой паузе. Так правило визита ложится в запрос целиком: одна кука,
-- один день, паузы не длиннее таймаута (CONTEXT.md, «Визит»).
--
-- Первое событие куки за день сравнивается само с собой — третий аргумент
-- lagInFrame. Паузы у него нет, шов ему не нужен, и номер первого визита
-- получается нулём: номера различают визиты внутри дня куки, а не считают их
-- от единицы.
--
-- Порядок в окнах полный — метка времени и event_id. Окна считаются двумя
-- разными сортировками (видно в EXPLAIN PIPELINE), и при равных метках они
-- вправе разойтись: шов достался бы одной строке, а нарастающий итог другой,
-- и сессия разъехалась бы надвое.
--
-- Тридцать минут написаны числом, а не приехали параметром: это правило
-- аналитики, а не константа мира. Генератор режет визиты тем же порогом;
-- разойдись эти два числа — увидит счётная сверка с VisitID
-- (docs/architecture/dds/session.md, «Сверка с эталоном»).
ALTER TABLE dds.session_stage_rep ON CLUSTER clickstream_cluster DROP PARTITION ALL; ALTER TABLE dds.session_stage_rep ON CLUSTER clickstream_cluster DROP PARTITION ALL;
INSERT INTO dds.session_stage_dist INSERT INTO dds.session_stage_dist
@@ -93,8 +78,8 @@ WITH
FROM seams FROM seams
) )
SELECT SELECT
-- Чекан визита: кука и его начало. Время входит в хеш секундой эпохи, -- Идентификатор зависит от посетителя и начала визита.
-- поэтому id держится за содержание сессии, а не за порядок сборки. -- Повторная сборка тех же событий дает тот же session_id.
cityHash64(client_id, min(event_time)), cityHash64(client_id, min(event_time)),
client_id, client_id,
event_date, event_date,
@@ -107,8 +92,10 @@ SELECT
countIf(event_type = 'purchase'), countIf(event_type = 'purchase'),
argMin(url, event_time), argMin(url, event_time),
argMax(url, event_time), argMax(url, event_time),
-- Атрибуция и паспорт куки берутся с первого события визита. У паспорта -- UTM, устройство и город берутся по минимальному event_time.
-- любое событие дало бы то же: устройство и город у куки постоянны. -- Как и у страниц выше, при равном времени выбор между событиями
-- не определен: event_id в эти argMin/argMax не входит.
-- Устройство и город на стенде постоянны для одной куки.
argMin(utm_source, event_time), argMin(utm_source, event_time),
argMin(utm_medium, event_time), argMin(utm_medium, event_time),
argMin(utm_campaign, event_time), argMin(utm_campaign, event_time),
@@ -120,12 +107,7 @@ SELECT
now64(3, 'UTC') now64(3, 'UTC')
FROM numbered FROM numbered
GROUP BY client_id, event_date, visit_no GROUP BY client_id, event_date, visit_no
-- Капкана parallel_distributed_insert_select здесь нет, и это выигрыш -- Источник и цель распределены по cityHash64(client_id): локальная вставка
-- ко-локации: ключ шардирования у источника и цели один — cityHash64 от -- сохраняет нужную раскладку, parallel_distributed_insert_select менять
-- куки, — поэтому визит ляжет туда же, где лежали его события, исполнись -- не требуется. Перед заменой партиций ждем доставки строк на шарды.
-- вставка хоть на инициаторе, хоть локально на шардах. Заказу приходится
-- называть ноль вслух: совпадение ключей у него случайное
-- (sql/dds/order_rebuild.sql). Здесь выбор сделан обратный и он тоже
-- осознанный: настройку не называть, чтобы не притворяться, будто её значение
-- что-то решает.
SETTINGS distributed_foreground_insert = 1; SETTINGS distributed_foreground_insert = 1;
+8 -8
View File
@@ -1,11 +1,11 @@
-- DDS: подмена одного дня сессий собранной партицией двойника. -- DDS: заменяем раздел session_rep за выбранный день готовым разделом
-- промежуточной session_stage_rep. Раздел таблицы называется партицией.
-- ON CLUSTER выполняет замену на локальных таблицах каждого шарда.
-- На одном шарде раздел меняется целиком; общей транзакции для кластера нет.
-- Повтор с теми же строками session_stage_rep не добавляет копий в session_rep.
-- --
-- Механика и доводы целиком — sql/dds/order_replace.sql: замена идёт по -- Дату из session_scope.sql подставляет Jinja: параметр ClickHouse в
-- локальным таблицам, шард за шардом, атомарность пошардовая, день -- ALTER ... PARTITION здесь не поддерживается. Поиск дней следующего
-- подставляет Jinja, потому что параметров ALTER … PARTITION не берёт. -- запуска не обнаружит неполный день, если часть его строк уже есть в DDS.
--
-- Разница одна, и она про отказ. Несобранный день заказа вернуло бы в работу
-- следующее окно; сессиям возвращать нечем, и упавшая замена оставляет дыру
-- до тех пор, пока её не найдёт счёт хвоста (sql/dds/session_scope.sql).
ALTER TABLE dds.session_rep ON CLUSTER clickstream_cluster ALTER TABLE dds.session_rep ON CLUSTER clickstream_cluster
REPLACE PARTITION '{{ params.day }}' FROM dds.session_stage_rep; REPLACE PARTITION '{{ params.day }}' FROM dds.session_stage_rep;
+16 -35
View File
@@ -1,50 +1,31 @@
-- 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.
-- Позиция мира сама по себе не подтверждает доставку событий.
-- --
-- Правый край — последний прожитый день, и знает его только позиция мира. -- Точки с запятой в конце нет: драйвер добавляет FORMAT к этому запросу.
-- События живой день от прожитого не отличают: и посреди дня 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, «Готовность источника»).
WITH WITH
(SELECT min(EventDate) FROM ods.event_dist) AS first_source_day, (SELECT min(EventDate) FROM ods.event_dist) AS first_source_day,
first_source_day + ({position:UInt16} - 1) AS last_lived_day, first_source_day + ({position:UInt16} - 1) AS last_lived_day,
-- Хвост считается от первой дыры, а не от последнего собранного дня: -- Ищем первую дату событий ODS, за которую еще нет сессий DDS.
-- замены партиций падают порознь, и счёт по максимуму собранного дня -- Поиск после max(session_date) пропустил бы более раннюю потерянную дату.
-- дыру позади себя не заметил бы (доводы целиком — sql/dds/order_scope.sql). -- Если за день есть хотя бы одна сессия, он уже не считается пропущенным.
-- У сессий это единственный способ дыру найти: дышащего окна, которое -- GLOBAL берет даты сессий со всех шардов.
-- пересобрало бы день заодно, здесь нет. -- minOrNull возвращает NULL, если пропусков нет.
--
-- GLOBAL обязателен: обе таблицы распределённые, и обычный NOT IN сверял
-- бы дни источника с локальной частью слоя на каждом шарде.
-- minOrNull отличает «дыр нет» от даты по умолчанию.
( (
SELECT minOrNull(day) SELECT minOrNull(day)
FROM (SELECT DISTINCT EventDate AS day FROM ods.event_dist) FROM (SELECT DISTINCT EventDate AS day FROM ods.event_dist)
WHERE day GLOBAL NOT IN (SELECT session_date FROM dds.session_dist) WHERE day GLOBAL NOT IN (SELECT session_date FROM dds.session_dist)
) AS first_missing_day ) AS first_missing_day
SELECT SELECT
-- Живой день в слое отсутствует всегда, и работой это не считается: -- Отсутствующий день попадет в сборку только после его завершения.
-- дырой он станет, только когда мир его проживёт.
if(first_missing_day <= last_lived_day, first_missing_day, NULL) AS first_day, if(first_missing_day <= last_lived_day, first_missing_day, NULL) AS first_day,
last_lived_day AS last_day last_lived_day AS last_day
-- FINAL не нужен: повтор доставки не двигает ни край оси, ни набор дней. -- FINAL не нужен: повтор доставки не меняет начало календаря и набор дат.
WHERE (SELECT count() FROM ods.event_dist) > 0 WHERE (SELECT count() FROM ods.event_dist) > 0
+12 -8
View File
@@ -1,7 +1,10 @@
-- DM: сборка выручки в донор перед заменой дневных партиций. -- DM: расчет выручки за дни от first_day до last_day включительно.
-- DROP PARTITION ALL очищает локальные таблицы на каждом узле; причина не -- DROP PARTITION ALL очищает промежуточную revenue_daily_stage_rep.
-- применять TRUNCATE разобрана в соседнем sql/dds/order_rebuild.sql. -- Затем 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 ALTER TABLE dm.revenue_daily_stage_rep ON CLUSTER clickstream_cluster
DROP PARTITION ALL; DROP PARTITION ALL;
@@ -22,8 +25,8 @@ SELECT
orders, orders,
units, units,
CAST(revenue, 'Decimal(18, 2)'), CAST(revenue, 'Decimal(18, 2)'),
-- Обычное деление Decimal усекает доли копейки вместо округления; -- Делим с шестью знаками после запятой, затем округляем до копеек.
-- запас масштаба сохраняет их до денежного результата. -- Так дробная часть сохраняется до округления среднего чека.
CAST( CAST(
round(divideDecimal(revenue, toDecimal128(orders, 0), 6), 2), round(divideDecimal(revenue, toDecimal128(orders, 0), 6), 2),
'Decimal(18, 2)' 'Decimal(18, 2)'
@@ -49,7 +52,8 @@ FROM
report_date, report_date,
product_category product_category
) )
-- Источник лежит по заказу, цель — по категории. Ноль заставляет инициатор -- Источник распределен по заказу, цель — по категории. Значение 0 оставляет
-- слить частичные агрегаты и разложить готовые строки по ключу цели. -- итоговую агрегацию и распределение по ключу цели на ноде, принявшей запрос.
-- distributed_foreground_insert = 1 ждет доставки строк перед заменой.
SETTINGS distributed_foreground_insert = 1, SETTINGS distributed_foreground_insert = 1,
parallel_distributed_insert_select = 0; parallel_distributed_insert_select = 0;
+7 -6
View File
@@ -1,8 +1,9 @@
-- DM: подмена одного дня выручки готовой партицией донора. -- DM: заменяем раздел revenue_daily_rep за выбранный день готовым разделом
-- -- промежуточной revenue_daily_stage_rep. Раздел таблицы называется партицией.
-- День приходит из результата revenue_daily_scope.sql, а не от пользователя. -- Дату из revenue_daily_scope.sql подставляет Jinja: параметр ClickHouse
-- Jinja нужна потому, что ClickHouse не подставляет параметр в PARTITION. -- в ALTER ... PARTITION здесь не поддерживается.
-- Замена идет на локальной таблице ON CLUSTER: граница атомарности и повтор -- ON CLUSTER выполняет замену на локальных таблицах каждого шарда.
-- разобраны на таком же шаге в sql/dds/order_replace.sql. -- На одном шарде раздел меняется целиком; общей транзакции для кластера нет.
-- Повтор с теми же строками промежуточной таблицы не добавляет копий в основной.
ALTER TABLE dm.revenue_daily_rep ON CLUSTER clickstream_cluster ALTER TABLE dm.revenue_daily_rep ON CLUSTER clickstream_cluster
REPLACE PARTITION '{{ params.day }}' FROM dm.revenue_daily_stage_rep; REPLACE PARTITION '{{ params.day }}' FROM dm.revenue_daily_stage_rep;
+10 -14
View File
@@ -1,14 +1,11 @@
-- DM: объем пересборки выручки. -- DM: диапазон пересборки выручки, first_day и last_day включительно.
-- Последний слепок ODS задает даты оформления заказов, которые еще могут
-- измениться. Саму выручку считаем по DDS; ODS нужен только для выбора дат.
-- Левая граница расширяется до первого дня с оплаченными заказами DDS,
-- которого нет в витрине. Так первый запуск охватывает историю выручки.
-- По наличию даты нельзя обнаружить частично заполненный день.
-- --
-- Последний слепок ODS называет дни, состояние которых источник еще может -- Точки с запятой в конце нет: драйвер добавляет FORMAT к этому запросу.
-- изменить. Сам агрегат читает только DDS; ODS здесь нужен лишь для границ.
-- Первый запуск и повтор после частичного отказа расширяют левую границу до
-- первого дня оплаченных заказов, которого еще нет в витрине.
-- Почему хвост начинается с первого пропуска и чем за это платит ODS,
-- разобрано в sql/dds/order_scope.sql.
--
-- Файл кончается без точки с запятой: даг читает его одним запросом и ждет
-- одну строку с границами.
WITH WITH
(SELECT max(snapshot_date) FROM ods.order_dist) AS last_snapshot, (SELECT max(snapshot_date) FROM ods.order_dist) AS last_snapshot,
( (
@@ -24,8 +21,7 @@ WITH
( (
SELECT minOrNull(report_date) SELECT minOrNull(report_date)
FROM (SELECT DISTINCT order_date AS report_date FROM dds.order_v WHERE status = 'paid') FROM (SELECT DISTINCT order_date AS report_date FROM dds.order_v WHERE status = 'paid')
-- Правая сторона тоже распределенная: GLOBAL собирает множество дней -- GLOBAL собирает список дней витрины со всех шардов.
-- на инициаторе и не превращает запрос в запрещенное двойное чтение.
WHERE report_date GLOBAL NOT IN ( WHERE report_date GLOBAL NOT IN (
SELECT report_date FROM dm.revenue_daily_dist SELECT report_date FROM dm.revenue_daily_dist
) )
@@ -33,7 +29,7 @@ WITH
SELECT SELECT
least(window_start, coalesce(first_missing_day, window_start)) AS first_day, least(window_start, coalesce(first_missing_day, window_start)) AS first_day,
window_end AS last_day window_end AS last_day
-- Без ODS неизвестно окно. Без готового DDS донор останется пустым, а замена -- Для выбора диапазона нужны заказы в ODS, для расчета — в DDS.
-- сотрет уже собранные дни выручки. Поэтому для прогона нужны оба источника. -- Эта проверка наличия строк не подтверждает полноту источников.
WHERE (SELECT count() FROM dds.order_v) > 0 WHERE (SELECT count() FROM dds.order_v) > 0
AND (SELECT count() FROM ods.order_dist) > 0 AND (SELECT count() FROM ods.order_dist) > 0