From 75f28fe99faf279ea1ef480ca59f35ab7ad91523 Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Sun, 13 Sep 2026 16:32:45 -0400 Subject: [PATCH] =?UTF-8?q?docs(ingestion):=20=D1=83=D0=BF=D1=80=D0=BE?= =?UTF-8?q?=D1=89=D0=B5=D0=BD=D1=8B=20=D0=BE=D0=B1=D1=8A=D1=8F=D1=81=D0=BD?= =?UTF-8?q?=D0=B5=D0=BD=D0=B8=D1=8F=20=D0=B8=D1=81=D1=82=D0=BE=D1=87=D0=BD?= =?UTF-8?q?=D0=B8=D0=BA=D0=BE=D0=B2=20=D0=B8=20=D0=BF=D1=80=D0=B8=D0=B5?= =?UTF-8?q?=D0=BC=D0=B0=20=D0=B4=D0=B0=D0=BD=D0=BD=D1=8B=D1=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Зачем: помочь начинающему инженеру проследить событие и версию заказа от генератора до хранилища, включая ошибки и повторные поступления. Что: обновлены вводная генератора, справочники приема и модели заказа, разделы хранилища и комментарии SQL и дага. Сохранены технические условия; исполняемый код, схемы и поведение стенда не изменены. Проверка: make lint, git diff --check, 142 локальные ссылки и справка CLI. SQL без комментариев и Python AST без строк документации совпали с исходными. Генерация данных и сброс состояния не выполнялись. Refs #250 --- dags/orders_ingest.py | 35 +-- docs/architecture/dds/order.md | 193 +++++------- docs/architecture/orders/ingestion.md | 339 ++++++++------------ docs/architecture/storage.md | 433 ++++++++++++-------------- generator/README.md | 159 +++++----- sql/ddl/20-ods-tables.sql | 95 ++---- sql/ddl/30-ods-views.sql | 110 ++----- sql/ddl/40-stg-views.sql | 32 +- sql/dds/order_rebuild.sql | 61 ++-- sql/dds/order_replace.sql | 36 +-- sql/dds/order_scope.sql | 53 +--- sql/ods/order_parse.sql | 73 ++--- sql/stg/orders_raw_load.sql | 18 +- 13 files changed, 663 insertions(+), 974 deletions(-) diff --git a/dags/orders_ingest.py b/dags/orders_ingest.py index 7f81363..60af770 100644 --- a/dags/orders_ingest.py +++ b/dags/orders_ingest.py @@ -1,18 +1,13 @@ -"""Приём заказов: пакетный забор слепка из Kafka в хранилище. +"""Пакетный прием заказов: Kafka → STG → ODS. -Путь Kafka → STG → ODS у заказов один и живёт одним дагом: сначала забор -порции в `stg`, затем разбор того же среза в типизированный ODS и в таблицу -брака. Расписания у дага нет: его дёргает тот, кто положил слепок в топик, — -шаговый даг модельного времени после проигрыша дня, — и ждёт конца. +Шаговый даг модельного времени отправляет слепок, запускает orders_ingest +и ждет его завершения. Собственного расписания у orders_ingest нет. +Сначала pull_batch сохраняет порцию сообщений в STG, затем parse_batch +разбирает тот же срез по _load_id в заказы и ошибки. -Контраст с приёмом событий и есть урок. События тянет матвью, навсегда -подписанная на чтеца: приём идёт сам, пока идёт поток. Слепок заказов -приезжает раз в модельный день целой выгрузкой, у которой есть начало и конец, -и забирает её запрос по команде. Push против pull — два режима на одном стенде, -каждый там, где ему место по природе источника. - -Решения и доводы целиком — ADR 0008 (забор) и ADR 0010 (версии в ODS); форма -перехода — docs/architecture/orders/ingestion.md. +События принимаются непрерывно через материализованные представления. +Заказы читаются одним запросом; полнота слепка этим не гарантируется. +Правила и ограничения — docs/architecture/orders/ingestion.md. """ from __future__ import annotations @@ -24,8 +19,7 @@ from airflow.sdk import dag START_DATE = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC) -# Запросы лежат файлами в sql/ репозитория, разложенные по слоям хранилища; -# сюда их монтирует compose — всем службам Airflow сразу. +# Compose монтирует sql/ по этому пути во все службы Airflow. SQL_ROOT = "/opt/airflow/sql" @@ -34,15 +28,14 @@ SQL_ROOT = "/opt/airflow/sql" schedule=None, start_date=START_DATE, is_paused_upon_creation=False, - # Чтец у топика один, и группа потребителей у него одна. Два прогона разом - # дрались бы за неё, а слепок разъехался бы по двум _load_id; второй - # прогон подождёт своей очереди. + # Один запуск за раз: параллельные чтения разделили бы порцию + # между разными _load_id. max_active_runs=1, template_searchpath=SQL_ROOT, tags=["заказы"], ) def orders_ingest(): - """Забрать приехавший слепок заказов и разложить его по слоям.""" + """Сохранить порцию заказов и разобрать ее в ODS.""" pull_batch = SQLExecuteQueryOperator( task_id="pull_batch", conn_id="clickhouse_default", @@ -59,8 +52,8 @@ def orders_ingest(): do_xcom_push=False, ) - # ClickHouse не даёт транзакции между двумя вставками parse_batch. Один - # task_id сохраняет их порядок и одну точку повтора для неизменного load_id. + # Две вставки parse_batch не образуют транзакцию. Повтор задачи + # выполняет обе над тем же срезом STG по _load_id. pull_batch >> parse_batch diff --git a/docs/architecture/dds/order.md b/docs/architecture/dds/order.md index 1216412..4b9c6d6 100644 --- a/docs/architecture/dds/order.md +++ b/docs/architecture/dds/order.md @@ -1,134 +1,113 @@ # Модель заказа в DDS -Учебный результат: менти проектирует бизнес-сущность поверх готового ODS — -выбирает зерно, состав строки и способ наполнения — и видит идемпотентность -трансформации, устроенную иначе, чем в приёме: заменой партиции, а не -версиями. +DDS готовит заказ для анализа: выбирает текущее состояние из ODS, +разбирает позиции и переводит время в часовой пояс счетчика. Этот слой +показывает, как собрать модель данных для витрин и повторить загрузку +без удвоения заказов. ## Зерно -Одна строка — заказ в текущем состоянии, ключ — `order_id`. Модель отвечает -на вопрос «что сейчас известно о заказе»; историю версий не хранит: -читателя у истории нет ни в одной витрине, а ODS её и не обещает — -промежуточные версии съедают фоновые слияния. «Дыхание» выручки внутри окна -K живёт на текущем состоянии само: пересборка дня меняет статусы заказов, и -число дня меняется вместе с ними. +Одна строка — один заказ в последнем известном состоянии, ключ — `order_id`. +История изменений здесь не хранится. ODS тоже не служит ее архивом: +промежуточные версии удаляются при фоновых слияниях. ## Состав строки -Деловые поля источника переезжают из `ods.order_v` без переименований — -бэкенд уже говорит в snake_case: `order_id`, `user_id`, `status`, деньги в -`Decimal` (`items_total`, `discount`, `delivery`, `total`). +Поля заказа приходят из `ods.order_v` с именами источника: `order_id`, +`user_id`, `status`, `items_total`, `discount`, `delivery`, `total`. +Денежные колонки хранят рубли с копейками в `Decimal(18, 2)`. -Времена `created_at` и `updated_at` пересчитываются при загрузке в пояс -счётчика: DDS обслуживает человека с дашбордом ([дока -хранилища](../storage.md), «Часовые пояса»). Там же один раз считается -**день заказа** — колонка `order_date`, по образцу `EventDate` у события: -суточные срезы группируются по готовой дате и пояса не упоминают. +`created_at` и `updated_at` переводятся из UTC в `Europe/Samara`, пояс +счетчика сайта. Из `created_at` в этом поясе вычисляется `order_date` — +день оформления заказа. Витрины группируют по готовой дате; +правило времени общее для [хранилища](../storage.md#часовые-пояса). -Позиции разобраны из сырой строки `items` в три типизированных массива -одной длины — артикул, количество, цена (`Decimal`). Это родная для -ClickHouse денормализация: позиция не живёт без заказа и собственного ключа -не имеет, а разворачивает массивы читатель — `ARRAY JOIN` в витрине -выручки, с обогащением категорией через `dictGet` -([мастер-спека](../../specs/2026-07-30-stand-v2-realism.md), раздел 7). Разбор -JSON происходит один раз, на границе ODS → DDS, куда его и положила спека -(раздел 2); записывать его параллельными колонками-массивами или `Nested` — -выбор DDL, не модели. +Строка `items` из ODS разбирается в три массива одной длины: + +| Колонка | Что хранит | +|---|---| +| `item_sku` | Артикулы товаров | +| `item_quantity` | Количество каждого товара, `UInt32` | +| `item_price` | Цены за единицу в рублях, `Decimal(18, 2)` | + +Элементы с одинаковым номером описывают одну позицию заказа. Витрина +выручки разворачивает их через `ARRAY JOIN` и получает категорию товара +из справочника через `dictGet`. Разбор JSON выполняется один раз +при загрузке DDS. + +ODS проверяет только то, что `items` — массив. Разбор в DDS не заменяет +проверку его содержимого: например, некорректное количество может стать +нулем. Проверку суммы позиций участник добавляет в +[первом задании ступени 5](../../labs/5-1-order-checks.md). ## Движок и наполнение -`dds.order_rep` — `ReplicatedMergeTree` без колонки версии: партиция — -`order_date` (день заказа неизменен), ключ сортировки — `order_id`. -`dds.order_dist` шардирует по `cityHash64(order_id)`, тем же выражением, что -ODS: это единственный устойчивый ключ сущности. +`dds.order_rep` — таблица на `ReplicatedMergeTree`, без колонки версии. +Партиция хранит один `order_date`, ключ сортировки — `order_id`. +Запись через `dds.order_dist` распределяет заказы по `cityHash64(order_id)`, +как в ODS. Для чтения служит `dds.order_v` с явным списком колонок +над `dds.order_dist`; `FINAL` в этом слое не нужен. -Идемпотентность держит замена партиции, а не версии. Сбор дня — батч после -прихода слепка: трансформация собирает партицию заново из `ods.order_v` и -атомарно подменяет. Поэтому каждая строка физической таблицы — ровно один -заказ, а чтению не нужен `FINAL`. Пока окно K живо, день «дышит» — статусы -заказов меняются, и партиция дня пересобирается очередным запуском; на выходе -из окна день замерзает. +Даг `dds_transform` пересобирает выбранные дни из `ods.order_v`: -Публичная поверхность — `dds.order_v`: обычное представление с явным списком -деловых и собственных служебных колонок над `dds.order_dist`, без `FINAL`. -Роли объектов и доступ следуют общей [конвенции](../storage.md), «Имена -объектов». +1. [`order_scope.sql`](../../../sql/dds/order_scope.sql) определяет + непрерывный отрезок `first_day`…`last_day`. Он включает дни заказов + из последнего слепка и начинается раньше, если в DDS отсутствует + день, который есть в ODS. +2. [`order_rebuild.sql`](../../../sql/dds/order_rebuild.sql) очищает + промежуточную таблицу `dds.order_stage_rep` и наполняет ее через + `dds.order_stage_dist`. Это донор: в нем готовится результат, + пока основная таблица остается доступной для чтения. +3. [`order_replace.sql`](../../../sql/dds/order_replace.sql) заменяет + каждую дневную партицию основной таблицы готовой партицией донора. -Механика запуска — где собирается партиция и как подменяется на кластере — -сюда не входит: это устройство [дневного конвейера](../etl.md). +Повтор при неизменном ODS дает те же деловые данные без дублей — +это идемпотентность загрузки. Пока заказы дня входят в семидневное окно +источника, их состояние может меняться, и день пересобирается. +За пределами окна новые состояния не приходят, но пропущенный день +можно восстановить из ODS. Границы пересборки берутся из данных; +число семь в SQL хранилища не повторяется. + +Замена атомарна внутри одного шарда: читатель видит прежнюю или новую +партицию целиком. Замены на двух шардах независимы, поэтому между ними +запрос через Distributed может увидеть смешанный результат. +Донор остается до следующей сборки и доступен для диагностики. +Порядок задач и повторы — в [описании дневного конвейера](../etl.md). ## Служебные колонки -Строка получает `_load_id` и `_load_ts` запуска трансформации, собравшего -её партицию, — по общей конвенции хранилища. При замене партиции они -одинаковы у всех строк дня и отвечают на вопрос «каким запуском и когда -собран этот день». +`_load_id` и `_load_ts` показывают, какой запуск DDS собрал строку +и когда он выполнил вставку. Они одинаковы у строк одной сборки. +При следующей пересборке эти метки обновятся, даже если заказ не изменился. -Ярлыки ODS (`snapshot_date`, `_load_id` приёма, `_load_ts` прибытия) в модель -не переезжают. Они описывают наблюдение и доставку строки в ODS. Модель DDS -хранит текущее деловое состояние заказа. В ODS `snapshot_date` задаёт версию и -называет последний слепок, в котором источник показал заказ -([ADR 0014](../../adr/0014-order-version-snapshot-date.md)). Дату -наблюдения и координаты доставки читатель спрашивает у ODS по -`order_id`. Правило слоя: -**служебные колонки — координаты собственной загрузки слоя; чужие координаты -спрашиваются у их слоя по ключу сущности**. +`snapshot_date` и метки приема ODS в DDS не переносятся: они описывают +получение сообщения источника. Их можно найти в ODS по `order_id`. +В каждом слое служебные колонки относятся к его собственной загрузке. -Происшествие качества данных расследуется цепочкой, где каждый шаг спрашивает -свой слой на его языке: по `_load_id` строки DDS видно, какой запуск собрал -день; по `order_id` ODS показывает физические версии и брак — там свой -`_load_id` приёма связывает годные строки, ошибки и данные STG; дальше читают -STG, пока данные живы. Окно расследования задано сроками хранения: сырые данные -— трое суток, брак — месяц ([дока хранилища](../storage.md)). +При поиске ошибки начните с `_load_id` в DDS: он указывает на запуск +сборки. По `order_id` найдите версии заказа в ODS, а по их `_load_id` — +соответствующие сырые сообщения в STG и ошибки того же приема. +В таблице ошибок `order_id` может быть не разобран, поэтому нужное +сообщение ищут по `raw` и координатам Kafka. +Сырые сообщения хранятся около трех реальных суток, ошибки — месяц; +уточнения к срокам — в [справочнике хранилища](../storage.md#срок-жизни-сырых-данных). ## Читатели -- Витрина выручки `dm.revenue_daily_v`: заказы в `paid`, `ARRAY JOIN` - позиций, категория через `dictGet` (спека, раздел 7). -- Сверка `dm.purchase_vs_orders_v`: `order_id`, `order_date`, `items_total`, - `status`; соединение — GLOBAL JOIN с обеими сторонами-подзапросами - ([исследование соединений](../../research/2026-08-22-distributed-join.md)). -- Сборка карты идентичностей: пара `order_id` ↔ `user_id`, источник — - `dds.order_v`. - -## Не входит - -- Механика запуска трансформации и замены партиций на кластере — - [дневной конвейер](../etl.md). -- Контур проверок качества (даг DQ, `dm.dq_summary`) — отдельный разговор - этапа 4. -- Инкрементальное наполнение вместо пересборки — техдолг #2. - -## Отклонённые варианты - -- **Одно динамическое представление вместо физической таблицы.** Рамка стенда — - показать модель данных, которую хранилище готовит потребителям; урок - «слой — контракт, а не копия» уже закреплён за `dds.event_v`, а разбор - позиций и пересчёт времён при каждом чтении собирали бы модель заново на - лету. Публичное `dds.order_v` поверх готовой физической модели этому доводу - не противоречит: оно ничего заново не вычисляет. -- **`ReplacingMergeTree` ещё раз.** Повторил бы механику ODS вторым разом: - физически хранил бы устаревшие состояния, вернул бы `FINAL` - и обёртку-представление — а урок версий менти уже отыграл. -- **Отдельная сущность `dds.order_item`.** Вторая таблица, второй вопрос - шардирования и `JOIN` там, где хватает `ARRAY JOIN`; нормализация из - реляционного мира, а не модель под колоночное хранилище. `JOIN` менти - увидит в сверке, работу с массивами — только здесь. -- **Зерно «версия заказа» (история).** Читателя нет; ODS историю не - обещает, а мир при нужде переигрывается генератором. -- **Перенос ярлыков ODS** — см. «Служебные колонки». -- **Без модели вовсе: витрины читают `ods.order_v` напрямую.** Разбор JSON - и пересчёт времён расползлись бы по витринам, а модель данных — то, ради - чего сущность и строится, — не была бы показана. +- `dm.revenue_daily_v` считает выручку по заказам со статусом `paid`, + разворачивает позиции и дополняет их категорией товара. +- `dm.purchase_vs_orders_v` сопоставляет заказ с покупкой на сайте + по `order_id`, используя `order_date`, `items_total` и `status`. +- `dds.identity_map` связывает идентификаторы посетителей с `user_id` + через покупки и заказы. ## Связанные решения -- [ADR 0010](../../adr/0010-order-versions-in-ods.md) — версии заказов в - ODS, устойчивая поверхность `ods.order_v`. -- [Приём заказов](../orders/ingestion.md) — граница строгого приёма и - четыре координаты строки ODS. -- [Проталкивание условий сквозь - представления](../../research/2026-08-22-view-predicate-pushdown.md) — - резолюция #128 той же карты: цена `FINAL` и цепочек представлений, фон, на - котором здесь выбрана физическая таблица. +- [ADR 0010](../../adr/0010-order-versions-in-ods.md) — версии заказов в ODS + и чтение через `ods.order_v`. +- [Прием заказов](../orders/ingestion.md) — проверки формата и выбор + актуальной версии. +- [Исследование соединений](../../research/2026-08-22-distributed-join.md) — + почему сверка распределенных источников использует `GLOBAL JOIN`. +- [Проталкивание условий сквозь представления](../../research/2026-08-22-view-predicate-pushdown.md) — + как фильтр запроса влияет на объем чтения ODS. diff --git a/docs/architecture/orders/ingestion.md b/docs/architecture/orders/ingestion.md index 52aaa21..acc9f39 100644 --- a/docs/architecture/orders/ingestion.md +++ b/docs/architecture/orders/ingestion.md @@ -1,241 +1,172 @@ -# Приём заказов из Kafka в ODS +# Прием заказов из Kafka в ODS -Учебный результат: менти различает версию бизнес-сущности, наблюдение источника -и запуск загрузки, а затем читает физические версии через явную поверхность -текущего состояния. - -## Проблема - -Заказы приезжают полным слепком окна изменяемости, но Kafka передаёт его -отдельными сообщениями и не сообщает потребителю, где слепок закончился. -Прямое чтение Kafka Engine возвращает одну порцию без признака полноты. -Поэтому `snapshot_date` не может служить границей публикации: `REPLACE -PARTITION` могла бы заменить партицию дня неполным набором строк. - -Одновременно типизированный ODS не должен принимать правдоподобные значения по -умолчанию из грязного JSON или останавливать весь пакет из-за одной строки. - -## Цели - -- сохранить пакетный забор как контраст потоковому приёму событий; -- один раз принять байты в STG и независимо разложить строки на годные и брак; -- хранить в ODS типизированные версии заказов, не публикуя слепок заменой - партиции; -- дать следующим слоям один корректный способ прочитать текущее состояние; -- оставить код проверки коротким и ограничить его контрактом провода. - -## Не входит - -- модель заказа в DDS: её зерно, связи, материализация и способ наполнения; -- проверка полей внутри `items`, переходов статуса и равенств денежных сумм; -- удаление заказа по отсутствию в следующем слепке; -- маркер конца слепка, опись ожидаемых строк и транзакция между целями ODS. +Здесь разобран путь одного заказа: сообщение Kafka сохраняется в STG, +проверяется и попадает либо в ODS, либо в таблицу ошибок. Следующие слои +читают из ODS последнюю принятую версию каждого заказа. ## Поток данных -После завершения генератора Airflow один раз читает байтовый Kafka-чтец и -записывает полученную порцию в `stg.orders_raw`. Даг зовётся `orders_ingest`, а -запускает его шаговый даг модельного времени, который положил слепок в топик и -ждёт конца прогона. Все строки получают `_load_id`, равный `run_id` Airflow. -`_load_ts` вычисляется при этой записи и дальше переносится без пересчёта. -Запрос забора лежит в -[`sql/stg/orders_raw_load.sql`](../../../sql/stg/orders_raw_load.sql): -`SQLExecuteQueryOperator` задаёт подключение и передаёт `run_id` как параметр -ClickHouse `load_id`, а преобразование остаётся в SQL своего слоя. +Источник отправляет в топик `orders` дневной слепок: по одному сообщению +на заказ. Слепок дня D содержит состояния заказов, оформленных в дни +D−6…D, с учетом [задержек первого появления](fate.md). +В начале модельного времени окно короче. Выгрузку за D отправляет +генератор при запуске дня D+1; подробности — в [описании слепка](snapshot.md). -Один следующий `task_id` отвечает за весь переход STG → ODS. Внутри него два -последовательных `INSERT SELECT` читают неизменный срез по `_load_id`: первый -пишет годные строки в `ods.order`, второй — брак в `ods.order_errors`. -Транзакции между запросами нет. При частичном сбое Airflow повторяет весь -`task_id`; одинаковые исходные строки и служебные метки не вычисляются заново. -Обе вставки лежат в одном самодостаточном файле -[`order_parse.sql`](../../../sql/ods/order_parse.sql). Airflow сначала -шаблонизирует файл, затем делит получившийся SQL на выражения и исполняет их по -порядку в одной задаче `parse_batch`. +После отправки шаговый даг модельного времени запускает +[`orders_ingest`](../../../dags/orders_ingest.py) и ждет его завершения. +У `orders_ingest` нет расписания; его два шага выполняются по порядку: -Условия запросов взаимодополняющие: один общий предикат определяет годную -строку, а ветвь брака использует его буквальное отрицание. Все функции предиката -возвращают результат без исключения, а сам предикат всегда заканчивается в -`true` или `false`, не в `NULL`. Постоянный классификатор между STG и ODS для -этого не нужен. Предикат объявлен локальным макросом Jinja внутри -`order_parse.sql` и разворачивается в оба выражения. Макрос убирает -дублирование, но не вводит отдельный фрагмент, который нельзя исполнить -самостоятельно. +1. `pull_batch` читает порцию из `stg.orders_raw_kafka` и записывает ее + через `stg.orders_raw_dist`. Содержимое сообщения остается строкой `raw`, + рядом сохраняются координаты Kafka, `_load_id` и `_load_ts`. +2. `parse_batch` читает сохраненные строки с этим `_load_id`. Первый + `INSERT SELECT` пишет годные записи в `ods.order_dist`, второй — + остальные в `ods.order_errors_dist`. + +`_load_id` равен `run_id` Airflow и связывает строки одного запуска. +`_load_ts` — реальное время записи в STG; при разборе оно переносится +без пересчета. Запросы лежат в +[`orders_raw_load.sql`](../../../sql/stg/orders_raw_load.sql) и +[`order_parse.sql`](../../../sql/ods/order_parse.sql). +`SQLExecuteQueryOperator` передает `run_id` параметром ClickHouse `load_id`. + +Две вставки разбора используют одно условие годности, записанное макросом +Jinja внутри `order_parse.sql`. Перед выполнением макрос подставляется +в оба запроса; ветвь брака берет отрицание этого условия. Оно возвращает +только истину или ложь и не вызывает исключений на некорректном `raw`. +Так каждая строка сохраненной порции получает ровно один результат. + +Транзакции между вставками нет: первая может завершиться, а вторая упасть. +Повтор `parse_batch` выполняет обе вставки над тем же срезом STG. +Текущее состояние заказа останется прежним, но таблица ошибок может +получить повторные строки с теми же `_load_id` и координатами Kafka. +Обе вставки ждут доставки строк на шарды и используют ключ целевой таблицы; +настройки объяснены в [справочнике хранилища](../storage.md#раскладка-по-шардам). ## Граница строгого приёма -Единица решения — одна строка `stg.orders_raw`. Корень должен быть -JSON-объектом с точным набором ключей: `order_id`, `user_id`, `status`, -`created_at`, `updated_at`, `items_total`, `discount`, `delivery`, `total`, -`items`, `snapshot_date`. +Проверяется одна строка `stg.orders_raw_dist`. Корень JSON должен быть +объектом с точным набором из одиннадцати ключей: +`order_id`, `user_id`, `status`, `created_at`, `updated_at`, +`items_total`, `discount`, `delivery`, `total`, `items`, `snapshot_date`. -Скалярные поля проверяются по типу JSON. Деньги дополнительно обязаны быть -строками неотрицательной суммы с ровно двумя знаками после точки — минус в эту -форму не входит; времена — строками RFC 3339 в UTC с обязательными -миллисекундами; дата слепка — строкой `YYYY-MM-DD`. `items` проверяется только -как JSON-массив. Каноническая форма и основания выбора зафиксированы в -[исследовании формата](../../research/2026-08-16-order-snapshot-wire-format.md). +| Поля | Условия приема | +|---|---| +| `order_id`, `status` | Строки JSON | +| `user_id` | Целое число JSON, которое разбирается в `UInt64` | +| `created_at`, `updated_at` | Строки UTC с миллисекундами, например `2026-06-01T12:34:56.000Z`; проверяются форма и разбор времени | +| `items_total`, `discount`, `delivery`, `total` | Неотрицательные суммы в рублях, записанные строками с двумя знаками после точки; проверяется разбор в `Decimal(18, 2)` | +| `snapshot_date` | Строка даты `YYYY-MM-DD`, которая проходит разбор | +| `items` | Массив JSON; сохраняется целиком в строковой колонке | -Проверять все верхнеуровневые поля здесь уместно: их одиннадцать, и десять -скалярных значений непосредственно образуют типизированную строку заказа. У -события из 47 полей проверяются только пять опорных; переносить то сокращение на -малый контракт заказа нет причины. Граница строгости заканчивается на форме -провода: содержимое позиций и бизнес-инварианты намеренно остаются ниже. +ODS проверяет формат верхнего уровня. Здесь нет проверки допустимых +статусов, их переходов, полей внутри `items` или равенства денежных сумм. +Позиции разбираются в [DDS](../dds/order.md); проверку их суммы участник +добавляет в [задании ступени 5](../../labs/5-1-order-checks.md). +Основания выбора формата — в [исследовании](../../research/2026-08-16-order-snapshot-wire-format.md). -Контракт источника задаёт частоту: **одной дате соответствует один слепок**. Это -свойство всего слепка, поэтому предикат одной строки его не проверяет. Вторая -выгрузка того же дня даст ничью по `snapshot_date`, и выбор строки потеряет -однозначность. Модельный источник выполняет условие; в настоящей интеграции -его фиксируют в договоре с источником -([ADR 0014](../../adr/0014-order-version-snapshot-date.md)). +Класс ошибки записывается в `error_class` по первому нарушенному условию: -Класс брака выбирается первым совпадением: +1. `not_an_object` — корень JSON не объект; +2. `keyset_mismatch` — набор ключей отличается от ожидаемого; +3. `field_invalid` — тип, форма или разбор значения не прошли проверку. -1. `not_an_object`; -2. `keyset_mismatch`; -3. `field_invalid`. - -Имя отдельного поля в класс не включается. В таблице ошибок остаются сырой -текст, метаданные доставки и `_load_id`, поэтому единичный случай можно разобрать -без постоянной детализации предиката. +Имя поля в класс не входит. Для разбора конкретной ошибки в +`ods.order_errors_dist` есть исходный `raw`, координаты Kafka и метки загрузки. ## Роль ODS -`ods.order_rep` хранит физически принятые версии в -`ReplacingMergeTree(snapshot_date)`. Ключ сортировки — `order_id`, партиция — -день неизменного `created_at`. `ods.order_dist` шардирует по -`cityHash64(order_id)`: только так все версии заказа попадают на один шард и -`FINAL` даёт корректный результат через распределённую таблицу. +`ods.order_rep` использует `ReplicatedReplacingMergeTree(snapshot_date)`. +Строки с одним ключом сортировки `order_id` считаются версиями одного заказа; +остается строка с максимальной `snapshot_date`. Партиция — +`toDate(created_at)` в UTC, ключ шардирования — `cityHash64(order_id)`. +Неизменный `created_at` и общий ключ направляют все версии заказа +в одну партицию и на один шард. -Четыре координаты отвечают на разные вопросы: +| Колонка | На какой вопрос отвечает | +|---|---| +| `snapshot_date` | На конец какого модельного дня источник показал это состояние? | +| `updated_at` | Когда запись заказа последний раз меняли в магазине? | +| `_load_id` | Какой запуск Airflow принял сообщение? | +| `_load_ts` | Когда сообщение записали в STG по реальным часам стенда? | -- `snapshot_date` — в слепке какого модельного дня источник показал строку; -- `updated_at` — когда строку последний раз меняли в источнике; -- `_load_id` — какой запуск Airflow принял строку; -- `_load_ts` — когда строка приехала в хранилище. +Например, заказ не менялся с 1 июня, но входит в слепки за 1 и 2 июня. +`updated_at` у строк одинаков, а `snapshot_date` различается. ODS оставит +строку за 2 июня вместе с метками ее доставки. Новая дата наблюдения +нужна DDS для определения окна пересборки, даже если состояние не изменилось. -Версию выбирает `snapshot_date`: выживает строка с максимальной датой слепка -([ADR 0014](../../adr/0014-order-version-snapshot-date.md)). Поэтому -`snapshot_date` показывает последнее наблюдение заказа. По строкам последней -даты слепка DDS определяет окно пересборки. `_load_id` и `_load_ts` принадлежат -той физической строке, которую сохранило слияние, и описывают её доставку. +Источник выпускает **один слепок на дату**. Повторная отправка дает то же +содержимое, а более поздний слепок не возвращает устаревшее состояние. +Эти условия не проверяются на одной строке при приеме. Если источник +пришлет два разных состояния с одной `snapshot_date`, эта колонка +не позволит определить, какое из них актуально. +Обоснование версии — [ADR 0014](../../adr/0014-order-version-snapshot-date.md). -Схема версий стоит на дисциплине источника: слепок с поздней датой не -показывает состояния старше. Модельный мир её держит — состояние читается на -границе суток, и слепок дня выходит чистой функцией зерна и дня; источник, -собирающий слепок из отставшей реплики, её ломает. Деловая метка `updated_at` -при этом монотонна по слепкам, поэтому выбор по дате слепка даёт тот же ответ, -что выбор по метке (замер — «Что проверено»). Тихая правка строки — ручным -`UPDATE` мимо приложения, миграцией — метку не двигает, но следующий слепок -перекроет её сам, пока заказ в окне изменяемости. За окном источник заказ -больше не показывает, и там помогает только сверка самих полей, в терминах -dbt snapshots — стратегия `check` вместо `timestamp`. +`ods.order_v` читает `ods.order_dist FINAL` и возвращает одну текущую строку +на `order_id`, даже если фоновые слияния еще не убрали старые версии. +Без `FINAL` физическая таблица может показать несколько строк заказа. +Историю ODS не гарантирует: старые версии постепенно удаляются при слияниях. +Отсутствие заказа в новом слепке не удаляет его из ODS. -Обычное чтение `_dist` показывает физически сохранившиеся версии и нужно для -диагностики. Их число зависит от фоновых слияний: ODS не служит архивом истории. -`ods.order_v` сохраняет те же источник-ориентированные поля без обогащения -данными модели, но возвращает одну актуальную версию на `order_id`. Сначала оно -может быть простым представлением над `_dist FINAL`; способ выбора можно -заменить, не меняя потребителей. - -Это представление остаётся ответственностью ODS: оно скрывает механику чтения -версий, но не строит бизнес-модель. DDS читает `ods.order_v` и отдельно решает, -какие сущности, связи и производные признаки ему нужны. Судьба служебных -колонок выше ODS решена вместе с моделью DDS: чужие ярлыки туда не переезжают -([модель заказа](../dds/order.md)). +DDS читает `ods.order_v`, переводит времена в пояс счетчика и разбирает +позиции. Для своей загрузки DDS записывает собственные служебные метки — +[модель заказа](../dds/order.md#служебные-колонки). ## Одно чтение Kafka -Стандартный слепок содержит около полутора тысяч строк, тогда как предел одной -порции на стенде — десятки тысяч сообщений. Поэтому один запуск Airflow делает -один прямой `SELECT`, без цикла до пустоты и без фиксации конечных офсетов. +Один запуск делает один прямой `SELECT`, без чтения в цикле до пустоты. +Обычный слепок содержит около полутора тысяч строк, а предел порции +на стенде — десятки тысяч сообщений. Это допущение о размере данных: +Kafka не сообщает запросу, что прочитан весь слепок. -Это допущение о размере стенда, а не доказательство полноты слепка. Если Kafka -вернёт короткую порцию, непрочитанный хвост останется в топике и приедет в один -из следующих запусков. После отказа от замены партиции это задержка, а не потеря -или публикация неполного дня. +Короткая порция оставляет непрочитанные сообщения для следующих запусков. +Поэтому ODS добавляет версии заказов по мере поступления. Замена партиции +по `snapshot_date` могла бы опубликовать неполный набор заказов +([ADR 0010](../../adr/0010-order-versions-in-ods.md)). -Отказ — случай другой, и «хвост дождётся» на него не распространяется. Офсеты -порции коммитятся в момент чтения, поэтому упавшая вставка уносит прочитанное с -собой: повторное чтение вернёт ноль, а часть строк может уже лежать на шарде. -Позиция мира при этом не двигается, и тот же день уедет заново — в STG он -окажется частичным дублем, а старый хвост, ушедший той же порцией, не вернётся -([ADR 0008](../../adr/0008-order-ingestion.md), «Следствия»). - -## Отклонённые варианты - -- Партиционная идемпотентность — `REPLACE PARTITION snapshot_date` или - `ReplacingMergeTree` по `(snapshot_date, order_id)`: у потребителя нет - признака полноты партиции, а дата наблюдения становится частью ключа - сущности. -- Обычный `MergeTree` в ODS с дедупликацией только в DDS: навсегда сохраняет - технические повторы там, где семантика версии уже известна. -- Маркер, опись, чтение до пустоты или конечные офсеты: добавляют протокол ради - объёма, который с большим запасом помещается в одну порцию. -- Два `task_id` или материализованный классификатор: дробят один короткий - переход слоя, не добавляя транзакционности. -- Представление текущего состояния в DDS и готовая схема `dds.order` в этой - задаче: перекладывают механику ODS на следующий слой и преждевременно задают - модель данных. - -## Риски и проверка - -- На малой управляемой порции дать по одной строке каждого класса брака и две - годные версии одного `order_id`. Две цели должны сохранить все непустые - сообщения, а `ods.order_v` — вернуть новую версию независимо от фонового - слияния. -- Повторить переход с тем же `_load_id`: строка в `ods.order_v` и её `_load_ts` - не должны измениться; версии одного заказа должны остаться на одном шарде и - в одной партиции. Таблица ошибок может снова записать тот же брак: совпавшие - `_load_id` и Kafka-координаты показывают повтор задачи. +Сбой после чтения опаснее короткой порции. Чтец подтверждает позиции +сообщений в Kafka — офсеты — при прямом чтении. Если последующая вставка +в STG упадет, эти сообщения уже не будут прочитаны той же группой повторно; +часть из них может успеть записаться на шарды. Позиция модельного времени +при неуспехе не продвигается. Повтор шагового дага снова отправит слепок +того же дня, но не восстановит старые сообщения других слепков, +если они попали в неудачную порцию. +Граница восстановления описана в [ADR 0008](../../adr/0008-order-ingestion.md#следствия). ## Что проверено -Переход STG → ODS снят на живом стенде 18 августа 2026 года при исполнении #94. -Опыты этого раздела прогнаны и подтвердили обещанное: разбиение строк STG на -годные строки и брак полное и непересекающееся; две версии одного заказа легли в -одну партицию и на один шард, а `ods.order_v` вернуло позднюю независимо от -фонового слияния; повтор задачи с тем же `_load_id` строку в `ods.order_v` и её -`_load_ts` не изменил, а таблица ошибок записала тот же брак второй раз; строки -опытов убраны. Числа, механика опытов и поведение ClickHouse, на которое всё это -опирается, — [документ хранилища](../storage.md), «Что проверено». +Это результаты прежних опытов на стенде; при редактуре они заново +не выполнялись. -После исправления -[#102](https://git.dementev.space/ddmitry/clickstream-data-platform/issues/102) -честный прогон дня снят на живом стенде 20 августа 2026 года: все 1694 строки -слепка дня 7 приняты, `ods.order_errors` осталась пустой. У заказа -`20260608-0007`, который прежде приходил с отрицательным итогом, теперь -`5490.00 − 823.50 + 299.00 = 4965.50`. +- 18 августа 2026 года (#93, #94) прямое чтение приняло слепок, + а разбор разделил строки на годные и ошибки без потерь и пересечений. + Две версии заказа попали на один шард в одну партицию; `ods.order_v` + вернул позднюю. Повтор разбора с тем же `_load_id` сохранил текущую строку + и ее `_load_ts`, но повторил ошибки. Подробности — + [справочник хранилища](../storage.md#что-проверено) и + [ADR 0008](../../adr/0008-order-ingestion.md#что-проверено). +- 20 августа после исправления + [#102](https://git.dementev.space/ddmitry/clickstream-data-platform/issues/102) + приняты все 1694 строки слепка дня 7 без ошибок. Итог заказа + `20260608-0007`: `5490.00 − 823.50 + 299.00 = 4965.50`. +- 21 августа (#124) пересборка через `SQLExecuteQueryOperator` завершила + `pull_batch` и `parse_batch`; в `system.query_log` осталось по одному + завершенному запросу `etl` на забор, годные строки и ошибки. +- 22 августа (#137) во всех 3034 парах с одинаковым `updated_at` + деловые поля совпали побайтово; по слепкам время изменения не убывало. +- 26 августа (#165) после двух слепков физическая таблица содержала + 3356 версий 1917 заказов. В `ods.order_v` у всех заказов была дата + последнего слепка, а деловые поля совпали с `argMax` по `updated_at` + из сырых данных. -При исполнении #124 21 августа 2026 года адресная пересборка мира прошла через -`SQLExecuteQueryOperator`: `pull_batch` и `parse_batch` завершились успешно. В -`system.query_log` на первой ноде осталось ровно по одному завершённому запросу -пользователя `etl` для забора в STG, записи годных строк и записи брака. - -Дисциплина источника замерена 22 августа 2026 года при исполнении #137. -Во всех 3034 парах строк с равным `updated_at` деловые поля совпали -побайтово, а по слепкам метка ни разу не убывала. - -Текущее состояние проверено на пересобранном мире 26 августа 2026 года при -исполнении #165. После двух слепков физическая таблица содержала 3356 версий -1917 заказов. У всех строк `ods.order_v` дата слепка совпала с последним слепком -заказа; деловые поля совпали с эталоном `argMax` по `updated_at` из сырых -данных. В обеих сверках расхождений нет. - -Забор из Kafka в STG снят на живом стенде 18 августа 2026 года при исполнении -#93: одно прямое чтение приносит весь слепок дня, метаданные доставки доступны, -офсеты коммитятся, а сбой забора не двигает позицию мира. Числа — [ADR -0008](../../adr/0008-order-ingestion.md), раздел «Что проверено». - -MCP Context7 в сессии проектирования был недоступен. На локальном ClickHouse -`26.3.17.56` проверено, что прямой `SELECT` Kafka Engine завершается после -одной порции, а `FINAL` через `Distributed` исполняется на таблицах шардов. -Поэтому версии одного `order_id` направляются на один шард. Фоновое схлопывание -`ReplacingMergeTree` и необходимость точного чтения сверены с -[официальной документацией](https://clickhouse.com/docs/reference/engines/table-engines/mergetree-family/replacingmergetree), -поведение чтения — с исходниками той же версии +На ClickHouse `26.3.17.56` отдельно проверены одна порция при прямом +`SELECT` Kafka и выполнение `FINAL` на шардах через Distributed. +При проектировании Context7 был недоступен; чтение сверяли с исходниками +той же версии: [`StorageKafka.cpp`](https://github.com/ClickHouse/ClickHouse/blob/v26.3.17.56-lts/src/Storages/Kafka/StorageKafka.cpp) и [`KafkaSource.cpp`](https://github.com/ClickHouse/ClickHouse/blob/v26.3.17.56-lts/src/Storages/Kafka/KafkaSource.cpp). +Повторная сверка правил версий с документацией — в +[справочнике хранилища](../storage.md#актуальные-события). ## Связанные решения diff --git a/docs/architecture/storage.md b/docs/architecture/storage.md index 51d8a9e..baa043d 100644 --- a/docs/architecture/storage.md +++ b/docs/architecture/storage.md @@ -122,112 +122,110 @@ DDS: представление читает события ODS. ## Служебные колонки -Собственные колонки не повторяют имён виртуальных. Виртуальные даёт движок: -`_topic`, `_partition`, `_offset`, `_timestamp` у Kafka, `_shard_num` у -`Distributed` и прочие. Если положить на диск колонку с таким же именем, в -матвью перестанет читаться, что дано движком, а что положено нами, — а это -ровно то различие, ради которого метаданные доставки и хранятся. Поэтому они -ложатся под именами `kafka_topic`, `kafka_partition`, `kafka_offset`, -`kafka_timestamp`, рядом — `consumer_host`, имя читавшей ноды: виртуальные -колонки его не несут, а после записи в `Distributed` он уже невосстановим. +STG сохраняет исходное сообщение в строке `raw` и добавляет сведения +о доставке. JSON на этом шаге не разбирается: некорректное содержимое +тоже попадает в таблицу. -Типы у них такие: `kafka_topic` и `consumer_host` — `LowCardinality(String)`, -значений мало и они повторяются; `kafka_partition` и `kafka_offset` — `UInt64`. -С `kafka_timestamp` сложнее, и форма его решена на стенде. Меток времени движок -даёт две: `_timestamp` — `Nullable(DateTime)`, то есть секунды, и -`_timestamp_ms` — `Nullable(DateTime64(3))`, миллисекунды. Колонка объявлена -`Nullable(DateTime64(3, 'UTC'))` и заполняется из `_timestamp_ms`: у брокера -метка миллисекундная, соседняя `_load_ts` тоже миллисекундная, а сырой слой -хранит приехавшее, и округлять ему нечего. Обнуляемость нужна отдельно от -разрядности: брокер метку заполняет не всегда, а необнуляемый тип значил бы либо -падение приёма на первом сообщении, либо тихие нули за 1970 год. +| Колонка STG | Что хранит | +|---|---| +| `kafka_topic` | Имя топика из виртуальной колонки Kafka `_topic` | +| `kafka_partition` | Номер партиции из `_partition` | +| `kafka_offset` | Позицию сообщения внутри партиции из `_offset` | +| `kafka_timestamp` | Метку времени Kafka из `_timestamp_ms`, если она есть | +| `consumer_host` | Имя ноды, прочитавшей Kafka | +| `_load_ts` | Реальное время записи сообщения в STG | +| `_load_id` | Только у заказов: `run_id` запуска Airflow, который принял порцию | -Заполняются все они выражением в `SELECT` матвью приёма, а не `DEFAULT` в -таблице. Для `consumer_host` это обязательно: `DEFAULT hostName()` вычисляется -на шарде-получателе, то есть назвал бы не ту ноду, которая читала топик, — -колонка молча отвечала бы на другой вопрос. +Топик, партиция и смещение вместе позволяют найти сообщение Kafka. +`kafka_topic` и `consumer_host` имеют тип `LowCardinality(String)`, +`kafka_partition` и `kafka_offset` — `UInt64`. +`kafka_timestamp` имеет тип `Nullable(DateTime64(3, 'UTC'))`: сохраняет +миллисекунды и допускает отсутствие метки. -Само сообщение лежит в колонке `raw` тем, чем пришло: чтец читает байты и ничего -не проверяет, поэтому там оказываются и целые сообщения, и мусор. События ниже -разбирают матвью ODS ([ADR 0005](../adr/0005-event-ingestion.md)), заказы — -пакетный шаг ([ADR 0008](../adr/0008-order-ingestion.md), [ADR -0010](../adr/0010-order-versions-in-ods.md)). +Виртуальные колонки предоставляет движок, а колонки STG хранят значения +на диске. Их имена различаются, чтобы не скрыть виртуальную колонку +собственной. Значения вычисляются в запросе приема. В частности, +`hostName()` должен выполняться на ноде, читающей Kafka: +`DEFAULT hostName()` только в локальной таблице назвал бы шард-получатель. -Движок таблицы STG — обычный `ReplicatedMergeTree`, `ORDER BY (kafka_partition, -kafka_offset)`: разбор полётов идёт от «какое сообщение», другого ключа у -таблицы STG и нет. Замену версий сюда ставить нельзя — она отменила бы свойство -слоя, ради которого он заведён: повтор доставки в STG обязан быть виден. +Таблицы STG используют `ReplicatedMergeTree` с +`ORDER BY (kafka_partition, kafka_offset)`. Повторная доставка остается +отдельной строкой; STG нужен в том числе для разбора таких повторов. -Метка времени загрузки зовётся `_load_ts`, тип `DateTime64(3, 'UTC')`. Ставится -она один раз при записи в STG: для событий — матвью приёма, для дневного слепка -заказов — пакетным шагом. Дальше метка переносится в ODS как есть и отвечает на -вопрос «когда строка приехала в хранилище», а не «когда её разобрали». В -`ods.event` она же служит колонкой версии `ReplacingMergeTree` и схлопывает -повтор доставки. У `ods.order` версию задаёт `snapshot_date` — дата слепка -источника; `_load_ts` только показывает, когда конкретная строка приехала. +`_load_ts` имеет тип `DateTime64(3, 'UTC')`. При переходе в ODS метка +копируется из STG без пересчета. У событий она также выбирает последнюю +доставку; у заказов версию задает `snapshot_date`. `_load_id` заказа +переносится в годную строку ODS и в таблицу ошибок. +У непрерывного потока событий нет запуска Airflow, поэтому `_load_id` +в его STG и ODS отсутствует. -`created_at` и `updated_at` заказа к служебным колонкам хранилища не относятся. -Они приезжают в сообщении как аудит строки в БД источника и в ODS разбираются в -`DateTime64(3, 'UTC')`; ClickHouse их не создаёт и добавляет рядом собственную -`_load_ts`. Совпадение слов «техническое время» не делает эти часы одной осью. +Время события `UTCEventTime` и дата `EventDate` приходят из счетчика сайта. +Поля заказа `created_at` и `updated_at` приходят из базы магазина. +Они относятся к модельному времени; `_load_ts` — к реальным часам стенда. +Например, событие июня можно загрузить в сентябре: его `EventDate` +останется июньской, а `_load_ts` будет сентябрьской. -Пакетной переобработки у событий в ODS нет: слой наполняют матвью, а не задание -Airflow. Переделать разобранное руками можно вставкой из STG с фильтром по -`_load_ts` в пределах трёхсуточного окна; ничья по версии разрешается в пользу -вставленного позже. Заказы разбирает пакетный шаг из неизменного среза STG по -`_load_id`; дневные партиции ODS он не заменяет. - -Имя согласовано с каноном служебных полей соседнего учебного стенда на -Greenplum, чтобы словарь был общим у двух хранилищ; ведущее подчёркивание у -технических колонок — распространённая запись, -её же используют Fivetran, Airbyte и Stitch. С правилом выше это не спорит: -запрещено совпадать с именами виртуальных колонок, а не носить подчёркивание. - -Общего идентификатора пачки загрузки (`_load_id`) нет. У потока событий нет ни -пачки, ни `run_id`, и колонка была бы пустой формальностью. У заказов читатель -назван: `_load_id` равен `run_id` Airflow и переносится из STG в годную строку -ODS и в таблицу ошибок. Нужен ли он выше ODS, решается вместе с моделью DDS. +События разбираются материализованными представлениями. Автоматической +пакетной переобработки у них нет; сохраненный срез STG можно разобрать +отдельным запросом до истечения срока хранения. При повторной вставке +с тем же ключом и `_load_ts` `ReplacingMergeTree` оставляет более поздно +вставленную строку. Заказы разбирает задача Airflow по `_load_id`. +В DDS метки загрузки уже относятся к собственной сборке слоя — +[пример заказа](dds/order.md#служебные-колонки). ## Актуальные события -`ods.event_rep` хранит физически принятые версии события в -`ReplacingMergeTree(_load_ts)`, а `ods.event_dist` служит для записи и -диагностики доставки. Обычное чтение этой пары может показать несколько строк -одного `WatchID`, пока фоновые слияния их не схлопнули. +Повторная отправка события может дать несколько строк в `ods.event_rep`. +Движок `ReplicatedReplacingMergeTree(_load_ts)` считает версиями строки +с одинаковым полным ключом сортировки: +`(CounterID, EventDate, intHash32(ClientID), WatchID)`. +При слиянии остается строка с максимальным `_load_ts`, то есть последняя +доставка. Один `WatchID` обозначает событие, но физический ключ замены шире. -`ods.event_v` — обычное представление над `ods.event_dist FINAL`. Оно сохраняет -имена, типы и зерно события, но всегда возвращает актуальную версию. Актуальное -состояние читают через него, а не повторяют `FINAL` в каждом запросе. -`dds.event_v` отвечает за другое: даёт событию язык бизнес-модели, в том -числе snake_case-имена и расшифровку кодов. +`ods.event_v` читает `ods.event_dist FINAL`: выбирает актуальную строку +при выполнении запроса, не дожидаясь фоновых слияний. Поэтому события +считают через `_v`; число строк в `_dist` зависит от уже прошедших слияний. +`dds.event_v` использует этот результат, дает колонкам имена в `snake_case` +и расшифровывает коды. -Типы ODS обычно повторяют контракт выгрузки. Денежный `purchaseRevenue` — -исключение: источник объявляет `Array(Float64)`, а материализованное -представление ODS разбирает числа сырого JSON сразу в `Array(Decimal(18, 2))`. -DDS наследует этот тип без нового преобразования. Прямой разбор подтвержден -документацией ClickHouse через Context7 и запросом к ClickHouse 26.3; сравнение -с сырыми данными подтвердило: копейки всех покупок целы. +Выбор по полному `ORDER BY` и максимальной версии, включая чтение с `FINAL`, +сверен через Context7 с [документацией ClickHouse](https://clickhouse.com/docs/reference/engines/table-engines/mergetree-family/replacingmergetree). +В стенде закреплена версия 26.3.17.56; `_v` использует `FINAL`, чтобы +результат не зависел от фоновых слияний. + +Типы ODS повторяют [описание выгрузки](../formats/clickstream-event.md) +с одним денежным исключением: `purchaseRevenue` объявлен источником +как `Array(Float64)`, а из JSON разбирается сразу в +`Array(Decimal(18, 2))`. Это сумма в рублях с копейками; DDS наследует тип. +`productPrice` содержит округленные целые рубли и не подходит для +восстановления точной суммы покупки. Прямой разбор `purchaseRevenue` +проверен на ClickHouse 26.3; результаты сохранены в разделе «Что проверено». ## Версии заказов +Слепки повторяют состояние заказов из семидневного окна источника. `ods.order_rep` хранит принятые версии в -`ReplacingMergeTree(snapshot_date)`: ключ сортировки — `order_id`, партиция — -`toDate(created_at)`. `ods.order_dist` шардирует по -`cityHash64(order_id)`. Все версии заказа лежат в одной партиции, чтобы их могли -схлопывать фоновые слияния, и на одном шарде, чтобы распределённый `FINAL` -выбрал одного победителя. +`ReplicatedReplacingMergeTree(snapshot_date)`: ключ сортировки — `order_id`, +партиция — `toDate(created_at)` в UTC. Запись через `ods.order_dist` +распределяет их по `cityHash64(order_id)`. У всех версий заказа одинаковы +день создания и шард, поэтому они могут заменять друг друга. -Физическая пара нужна для загрузки и диагностики. Обычное чтение показывает -версии, которые ещё не убрали фоновые слияния, и не является архивом истории. -`ods.order_v` служит поверхностью точного текущего состояния для следующих -слоёв. Представление сохраняет язык источника и не решает, какой станет модель -DDS. `snapshot_date` в нём одновременно дата наблюдения и колонка версии. Она -не доказывает полноту принятой порции, поэтому ODS не публикует партиции по -дате слепка. Модель версионного ODS записана в -[ADR 0010](../adr/0010-order-versions-in-ods.md), выбор колонки версии описан в -[ADR 0014](../adr/0014-order-version-snapshot-date.md), пакетный приём описан в -[спецификации приёма заказов](orders/ingestion.md). Контракт следующего слоя -задан в [модели заказа в DDS](dds/order.md). +`ods.order_v` читает `_dist FINAL` и оставляет строку с максимальной датой +слепка. Дата `snapshot_date` показывает последнее принятое наблюдение, +а `updated_at` — время изменения заказа в магазине. У неизменившегося +заказа в следующем слепке дата наблюдения обновится, время изменения — нет. + +Источник выпускает один слепок на дату, а поздний слепок показывает +не менее новое состояние. Отсутствие заказа в следующем слепке +не означает удаления. Физическая таблица не служит архивом: +старые версии исчезают при фоновых слияниях. + +Прием добавляет версии отдельными строками. `snapshot_date` не доказывает, +что Kafka отдала весь слепок, поэтому партиция по этой дате не заменяется. +Подробности — [прием заказов](orders/ingestion.md), +[ADR 0010](../adr/0010-order-versions-in-ods.md) и +[ADR 0014](../adr/0014-order-version-snapshot-date.md). +Следующий слой описан в [модели заказа DDS](dds/order.md). ## Часовые пояса @@ -289,169 +287,148 @@ UTC+4), и пересчёт идёт один раз при наполнении ## Приём: поток и его свойства -Раздел — про события. У заказов приём устроен иначе: слепок забирает по команде -даг `orders_ingest`, а не подписанная навсегда матвью — [ADR -0008](../adr/0008-order-ingestion.md) и [спецификация приёма -заказов](orders/ingestion.md). +События принимаются непрерывно. Заказы читает отдельным запросом +даг `orders_ingest` — [описание пакетного приема](orders/ingestion.md). +У событий путь такой: -Цепочка одна: чтец топика → матвью → STG → матвью разбора → событие и таблица -ошибок ODS. +```text +stg.hits_raw_kafka → stg.hits_raw_mv → stg.hits_raw_dist + ├→ ods.event_mv → ods.event_dist + └→ ods.event_errors_mv → ods.event_errors_dist +``` -Чтец стоит на обеих нодах и читает одной группой потребителей — имя группы -`clickstream_hits`, и оно одинаково на обеих нодах по построению: DDL идёт -`ON CLUSTER` и макросов в имени не содержит. Разные группы дали бы каждой ноде -полную копию топика, и это отдельная сцена для лабы, а не рабочий режим. Имя -кластера в `ON CLUSTER` и в движке `Distributed` — `clickstream_cluster`, оно -задано в `infra/clickhouse/config.d/cluster.xml`. +Таблицы Kafka на обеих нодах используют одну группу `clickstream_hits`. +Ноды делят чтение партиций топика. Разные группы прочитали бы каждая +свою полную копию потока. Имя группы одинаково в DDL на всем +`clickstream_cluster`; конфигурация кластера — +`infra/clickhouse/config.d/cluster.xml`. -**Источник матвью разбора — `stg.hits_raw_dist`, а не локальная таблица.** -У матвью две привязки: источник, на вставку в который она срабатывает, и цель, -куда пишет. Распределённая таблица — лицо слоя, локальная — его хранилище; -потребитель слоя цепляется к лицу. Практически это значит, что разбор идёт на -той же ноде, что читала Kafka, в момент вставки первой матвью — до раскладки по -шардам. +Материализованные представления разбора срабатывают на вставку +в `stg.hits_raw_dist`, до распределения строк по шардам. +Оба видят вставляемую порцию на той ноде, которая читает Kafka. +`stg.hits_raw_mv` создается после них: его подключение запускает +чтение топика. Если начать чтение раньше, сообщения успеют попасть +в STG без разбора в ODS. -**Вставка фоновая, и окно потери мы принимаем.** Вставка в распределённую -таблицу кладёт блок в локальный спул и сразу возвращает управление, а Kafka -коммитит офсеты по факту работы матвью — то есть по факту записи в спул. Топик -уже считает сообщение прочитанным, хотя на шарде его ещё нет: умри нода в этом -промежутке — сообщения не перечитаются. +Условие приема проверяет три вещи: -Закрывает окно настройка `distributed_foreground_insert = 1`, и на ETL-вставках -Airflow она стоит — там это обычный `SETTINGS` у запроса. На пути приёма её нет, -и по трём причинам. Вставку выполняет фоновый поток Kafka-движка, своего запроса -у него не бывает, так что настройка уровня запроса доехала бы только профилем -пользователя в конфигурации ноды. Синхронный режим связывает шарды: пока второй -недоступен, вставка падает, офсеты не коммитятся, и приём встаёт целиком — тогда -как при фоновом первая нода продолжает принимать и копит спул для соседа. -Платится при этом не одно ожидание на блок, а три распределённые вставки — STG, -событие, ошибки, — и все внутри потока-потребителя, что само по себе повод для -ребаланса по таймауту сессии. Против всего этого — окно в сотню миллисекунд на -ноутбуке, где мир пересобирается одной командой. Размен не в пользу настройки, а -компромисс полезнее показать, чем спрятать за галочкой. +- корень `raw` — объект JSON; +- набор верхнеуровневых ключей совпадает с 47 полями формата; +- пять опорных полей, `WatchID`, `VisitID`, `ClientID`, `EventDate` + и `UTCEventTime`, разбираются в ожидаемые типы. -Ещё одно место, где сырой слой хранит не всё приехавшее: запись с пустым -значением и запись-надгробие проходят молча, не оставляя строки. Свойство -измерено и принято осознанно — подробности в разделе «Что проверено». +Остальные поля извлекаются по описанию выгрузки без отдельных проверок. +Это граница приема: годная строка не означает, что проверен смысл +всех 47 значений. Подробности — в [ADR 0005](../adr/0005-event-ingestion.md). -**Гарантии нет ни в одну сторону — есть два узких окна.** Окно потери описано -выше: нода умерла между коммитом офсетов и сбросом спула. Окно дубля -противоположное: нода умерла после записи на шард, но до коммита офсетов, и при -перечитывании сообщение приедет второй раз. Сказать про такой приём «хотя бы -один раз» нельзя — это обещало бы, что потерь не бывает, а они возможны. +`ods.event_mv` принимает строки, для которых условие истинно, +`ods.event_errors_mv` — для которых оно ложно. Условие не должно +возвращать `NULL`: такую строку не выбрали бы ни оно, ни его отрицание. +Разбор использует функции без исключений на некорректном сообщении, +поскольку ошибка материализованного представления прерывает вставку +и останавливает чтение Kafka до устранения причины. -Дубль ниже по течению ведёт себя по-разному. В ODS его схлопнет -`ReplacingMergeTree`, а таблицы STG вовсе не устраняют дубли: перезаливка -модельного дня честно удваивает `count()` в STG, и живёт эта пара до истечения -срока хранения. Это свойство слоя, а не поломка, — но обещание идемпотентности -конвейера к сырому слою не относится. +### Доставка и повторы -Оговорка к последнему: у семейства `Replicated*` есть своя дедупликация — блок с -тем же хешем, вставленный повторно, отбрасывается (`insert_deduplicate`). -Удвоение строк STG проходит мимо неё только потому, что при повторном чтении -`_load_ts` новый и хеш блока другой. Свойство слоя держится на этом, а не на -отсутствии механизма. +Вставка в Distributed на этом пути фоновая: блок сначала записывается +в локальную очередь на диске, затем отправляется на шарды. +Kafka подтверждает обработку после выполнения материализованных +представлений, когда данные еще могут находиться в очереди. +Если после подтверждения потерять ноду вместе с очередью, сообщения +не перечитаются. Возможен и повтор: данные уже записались на шард, +но подтверждение Kafka не успело сохраниться. +Поэтому прием не гарантирует доставку каждого сообщения хотя бы один раз. -**Матвью разбора две, и их условия обязаны делить поток без зазора и без -нахлёста.** Одна забирает годные строки в `ods.event_dist`, вторая — брак в -`ods.event_errors_dist`. Строка, подошедшая обеим, задвоится; не подошедшая ни -одной — исчезнет молча. Держится это формой: второе условие пишется буквальным -отрицанием первого, а сам предикат NULL не возвращает ни в одной своей части, — -иначе трёхзначная логика даст строку, которую не возьмёт ни `условие`, ни -`NOT условие`. Обнуляемый разбор в предикате поэтому есть, но заканчивается -`IS NOT NULL`, а сравнения дают 0 или 1. Функции предиката не должны и бросать -исключений: упавшая матвью -роняет вставку и останавливает потребление до починки -([ADR 0005](../adr/0005-event-ingestion.md)). +В ETL-запросах Airflow `distributed_foreground_insert = 1` дожидается +записи на шарды. У фонового приема событий этой настройки нет: +недоступность одного шарда тогда остановила бы весь прием. В принятом +режиме доступная нода продолжает принимать и накапливает очередь +для недоступного шарда. Ограничение сохранено для учебного стенда; +подробности и наблюдения — в разделе «Что проверено». -Постоянной сверки **слоёв между собой** при этом нет и не должно быть. У сырого -слоя срок жизни трое суток, а ODS хранит всё, поэтому равенство «строки STG = -события + ошибки» разъедется само: строки STG истекут раньше, а перезаливка -модельного дня задвоит их, тогда как в ODS тот же повтор схлопнется. Правило, -красное в норме, учит не смотреть на оповещения ([ADR -0002](../adr/0002-monitoring-scope.md)). Равенство проверено разовым опытом при -исполнении #43, на управляемой пачке: отправили N сообщений — получили N строк -STG и N в сумме событий и ошибок. Постоянной целью такой опыт не становится, и -почему — в [карте проверок](testing.md), раздел «Интеграционная проверка -постоянной целью не становится». Счёт актуальных событий идёт через -`ods.event_v`, которое скрывает `FINAL`: голый `count()` по `ReplacingMergeTree` -зависит от того, сколько мержей успело пройти, и спека это прямо запрещает -(раздел 6). +В STG повтор сохраняется, в ODS его устраняет выбор актуальной версии. +Хотя `Replicated*` умеет отбрасывать повторно вставленные одинаковые блоки, +при новом чтении Kafka меняется `_load_ts`, а вместе с ним и хеш блока. +Поэтому повторная отправка модельного дня увеличивает число строк STG. -**Постоянная сверка одна, и она другого рода** — не слой против слоя, а ODS -против [паспорта мира](../../data/world-inventory.json), лежащего в git (`make -check-clickhouse`, пришла с #42). Разъехаться сама она не может, и в этом вся -разница: паспорт описывает восемь дней стартового мира, счёт обрамлён их датами, -повтор заливки схлопывается в `ods.event_v`, а срок хранения ей не помеха — ODS -хранит всё. Обе стороны равенства зафиксированы: одна кодом генератора, другая -файлом в git. У межслойной сверки такой опоры нет ни с одной стороны. +Пустое значение сообщения и значение `null` не дают строки даже в STG: +чтец продвигает позицию Kafka, но не передает строку на разбор. +Генератор таких сообщений не создает. Поведение измерено на стенде +и описано в разделе «Что проверено». + +### Что можно сверять + +Для короткого неизменного среза доставки можно проверить баланс: +число строк STG равно сумме годных строк и ошибок до удаления повторов. +Постоянно сравнивать полные таблицы так нельзя: STG удаляет старые данные, +ODS оставляет события и объединяет повторные доставки. +Счетчики по слоям тогда расходятся при штатной работе. + +`make check-clickhouse` сравнивает актуальные события в `ods.event_v` +с [паспортом мира](../../data/world-inventory.json) только за восемь +начальных дней. Эта проверка учитывает повторную отправку через `FINAL` +и не зависит от срока жизни STG. Границы остальных проверок — +в [карте проверок](testing.md). ## Срок жизни сырых данных -Сырые данные в STG живут трое суток реального времени и уходят сами. Трое — это -окно отладки: столько сырые данные хранятся на ноутбуке, чтобы менти успел -разобрать полёты, после чего перестаёт занимать место. Нарезка — по дню -загрузки, срок — по той же колонке `_load_ts`, снятие — целыми кусками -(`ttl_only_drop_parts`). В DDL значение проставлено явно, чтобы поведение не -зависело от умолчания версии. +TTL — правило автоматического удаления старых данных. В обеих таблицах +STG срок задан как три суток от `_load_ts`, по реальному времени загрузки. +Партиции тоже нарезаны по дню `_load_ts`. Это оставляет короткий период +для поиска ошибок и повторного разбора, после которого данные освобождают +место на диске. -Нарезать сырые данные по модельному дню события было бы соблазнительно — он -единица переобработки и он же ключ партиции в ODS, — но чистку это ломает. Кусок -снимается целиком, только когда в нём истекли все строки, а в партиции -модельного дня лежит приехавшее в разное реальное время: мерж склеит куски -разного возраста, самая свежая строка удержит весь кусок, и данные переживут -срок неограниченно. +Настройка `ttl_only_drop_parts = 1` удаляет только целые части данных — +файлы, которые ClickHouse создает при вставках и объединяет при слияниях. +Часть удаляется, когда срок истек у всех ее строк. Внутри партиции +дня загрузки возраст строк отличается не более чем на сутки. +Поэтому к трем суткам может добавиться до суток ожидания самой свежей +строки части, а затем ожидание фоновой обработки TTL. Точного времени +удаления правило не обещает. -В партиции дня загрузки склейка идёт точно так же, и строки в ней тоже разного -возраста — но не более чем на сутки, потому что партицию закрывает календарный -день. Отсюда и оценка: сырые данные живут трое суток плюс хвост до суток, а не -ровно трое. Оговорка про сроки: TTL исполняется на мержах, а не по будильнику, -так что «уходит само» здесь обещано, а «уходит вовремя» — нет. +Модельная дата для TTL не подходит: события июня могут загружаться +осенью, и срок от `EventDate` истек бы сразу. Партиция по модельному дню +тоже смешала бы доставки разного возраста: очередная отправка дня могла бы +продлевать хранение старых строк после слияния частей. -Модельного дня среди колонок STG нет вовсе. Он свойство содержимого, а -содержимое разбирает ODS — там `EventDate` и живёт, ключом партиции. Сырые -данные режутся по своим координатам: и разбор полётов, и переобработка фильтруют -по `_load_ts`, попадая в ключ партиции, а не идя сплошным проходом. Фильтр по -модельному дню вдобавок пропускал бы битые строки — у них дата не извлекается. - -Две оси времени тут не совпадают намеренно. Ось модельного времени начинается в -D0 и к реальному календарю не привязана; пакетный режим проигрывает две недели -модельного мира за минуты реальных. Поэтому у переобработки и у гигиены диска -разные часы, и обслуживают их разные средства. Декларативный TTL по модельной -дате был бы просто сломан: он отсчитывает срок от реального «сейчас» и удалял бы -эталонные дни прямо на входе. - -В бою сырой слой иногда собирают на движке `Null` — тогда он не хранится вовсе. -Такой вариант отвергнут: на стенде сырые данные нужны для отладки, поэтому окно, -а не ноль. +Для разбора сохраненных сообщений выбирайте небольшой срез по `_load_ts`. +Так запрос ограничивает партиции и сохраняет некорректные сообщения, +из которых модельную дату извлечь нельзя. `EventDate` появляется +отдельной колонкой только в ODS. В [задании ступени 6](../labs/6-1-event-parsing.md) +такой срез разбирается рядом с действующим приемом. ## Таблица ошибок -`ods.event_errors` держит строки, не прошедшие строгий приём, вместе с их сырым -текстом, метаданными доставки и классом брака. Ключи её собственные, потому что -у брака нет разобранных полей: шардируется `cityHash64` сырой строки — -`ClientID` у строки, которая не разобралась, взять неоткуда; нарезается по дню -загрузки, как и сырые данные; живёт месяц. Дольше сырых данных — намеренно: если -брак истекает вместе с ними, разбираться к моменту разбирательства будет уже -нечем. Снятие — целыми кусками (`ttl_only_drop_parts`), как у STG и по той же -причине: строки в партиции дня загрузки разного возраста не более чем на сутки, -и куску незачем переживать срок из-за самой свежей строки. +`ods.event_errors_dist` содержит сообщения, отвергнутые разбором событий: +исходный `raw`, сведения о доставке и класс ошибки `error_class`. +Класс имеет тип `LowCardinality(String)` и выбирается по первому +нарушенному условию: -Класс брака лежит в колонке `error_class` типа `LowCardinality(String)`. Без неё -в таблице копятся строки «что-то не так» без ответа на «что именно», а витрине -качества не на что опереться. Сами классы, их порядок и довод, почему порядок -обязателен, — в [ADR 0005](../adr/0005-event-ingestion.md). +1. `not_an_object` — корень JSON не объект; +2. `keyset_mismatch` — набор ключей не совпал с форматом; +3. `key_field_unparsed` — одно из пяти опорных полей не разобралось. -Движок — обычный `ReplicatedMergeTree`, без замены версий: схлопывать брак не по -чему, у него нет ключа сущности. `ORDER BY` — `(error_class, kafka_partition, -kafka_offset)`: смотрят такую таблицу от класса, а внутри класса — по координатам -доставки. +Порядок важен: строка вместо объекта нарушает сразу несколько условий, +но получает один класс. Полные условия — в +[ADR 0005](../adr/0005-event-ingestion.md). -`ods.order_errors` держит тот же диагностический минимум и `_load_id` -запуска. У заказов три класса по приоритету: `not_an_object`, -`keyset_mismatch`, `field_invalid`. Сырой текст остаётся рядом, поэтому класс не -разрастается до имени отдельного поля. Точная граница приёма — в -[спецификации заказов](orders/ingestion.md). +`ods.order_errors_dist` хранит такой же набор сведений и `_load_id` +запуска приема. Первые два класса совпадают с событиями, +третий называется `field_invalid`: у заказа проверяются все поля +верхнего уровня. Точные правила — в [описании приема заказов](orders/ingestion.md#граница-строгого-приёма). + +Таблицы ошибок используют `ReplicatedMergeTree` без замены версий: +у некорректного сообщения может не быть разобранного ключа сущности. +Шард выбирается по `cityHash64(raw)`, ключ сортировки — +`(error_class, kafka_partition, kafka_offset)`, партиция — день `_load_ts`. +Начать поиск можно с класса ошибки, затем перейти к сообщениям +по координатам Kafka. + +Срок хранения — один месяц от `_load_ts`, удаление целыми частями +с `ttl_only_drop_parts = 1`, как в STG. Ошибка вместе с исходным текстом +остается доступна дольше остальных сырых сообщений. Повторная обработка +может добавить тот же брак еще раз; у заказов повтор виден по совпадению +`_load_id` и координат Kafka. ## Справочники diff --git a/generator/README.md b/generator/README.md index 5e4699e..7731615 100644 --- a/generator/README.md +++ b/generator/README.md @@ -1,98 +1,95 @@ -# Генератор кликстрима +# Генератор событий и заказов -Клиентская сторона стенда: отсюда берётся поток событий — широкое событие по -образцу облачной выгрузки Яндекс Метрики. Устройство и принятые решения — -спека [«Генератор (этап 2)»](../docs/specs/2026-08-01-generator.md). +Генератор создает данные двух источников интернет-магазина. Счетчик сайта +отправляет в топик Kafka `hits` события: просмотр страницы, добавление товара +в корзину и покупку. Серверная часть магазина выгружает в `orders` дневной +слепок — состояние заказов, оформленных за последние семь модельных дней. +Оба источника работают; хранилище принимает события непрерывно, заказы — +по команде Airflow. -События уже есть и уже уезжают: день-функция отдаёт по паре (зерно, D) -упорядоченный поток трёх видов — просмотр страницы, корзина, покупка, — -канонический сериализатор превращает его в JSON, а проигрыватель гонит в файл -или в Kafka. Клиентская сторона на этом целая. Второй источник строится: -день отдаёт и заказы бэкенда — проекцию своих покупок с деньгами магазина, — -а судьба заказа и отправка слепка идут следующими тикетами. +Событие описывает отдельное действие посетителя. Заказ после оформления +может измениться, например перейти в статус `paid` или `cancelled`. +Поэтому один `order_id` встречается в нескольких слепках. Часть заказов +впервые появляется с задержкой в один или два дня; конец семидневного окна +от этого не сдвигается. Правила изменений и задержек — в +[описании судьбы заказа](../docs/architecture/orders/fate.md). + +Слепок снимается на конец модельного дня и отправляется на следующий день. +Команда `snapshot --day 1` отправляет слепок дня 0, а `snapshot --day 0` +не отправляет ничего: до начала модели заказов нет. События и слепки +отправляются отдельными запусками генератора. ## Как это работает -Корневое зерно ветвится на подпотоки, из них складываются когорты, аудитория -дня, визиты, строки событий и заказы. План поднимает состав мира по требованию, -а день-функция проживает один день без накопленного состояния. Код этой -цепочки, ее схема и запуск D0 собраны в -[путеводителе](../docs/architecture/generator/README.md). +Зерно генерации задает случайные выборы: какие посетители придут, +какие страницы откроют и что купят. При неизменных коде, каталоге +и настройках одинаковые зерно и номер дня дают одинаковые данные. +День можно рассчитать без генерации всех предыдущих дней. + +Сначала рассчитываются посещения и покупки. Из покупки получаются событие +`purchase` со стороны сайта и заказ со стороны магазина. Номер связывает +их: `purchaseID` в событии соответствует `order_id` заказа. Суммы могут +различаться: магазин учитывает скидку, доставку и удаленные позиции. +Деньги генератор считает целыми копейками; в JSON заказа это строки +в рублях с двумя знаками после точки, например `"1289.99"`. + +Готовые записи превращаются в JSON и отправляются в файл или Kafka: +одна запись занимает одну строку файла или одно сообщение. Событие содержит +47 полей по [описанию выгрузки](../docs/formats/clickstream-event.md), +запись заказа — поля [формата слепка](../docs/architecture/orders/snapshot.md#запись-на-проводе). +Подробное устройство генератора разобрано в +[техническом справочнике](../docs/architecture/generator/README.md). ## Что где лежит -- `src/clickstream_generator/world.py` — конфигурация мира: все его числа - одним местом. Правка любого — смена мира; крутить их и предлагается. -- `src/clickstream_generator/seeds.py` — иерархия зёрен: кто из какого - подпотока берёт случайность. На ней держится весь детерминизм. -- `src/clickstream_generator/plan.py` — план состава: кто есть в мире в - день D. Когорты, приток, двухкуковые пары, паспорта кук, человек за кукой - и счётчики — до генерации событий. -- `src/clickstream_generator/weights.py` — выбор по целым весам: один приём - на весь генератор, чтобы дисциплина целочисленной случайности не жила - копиями. -- `src/clickstream_generator/reference.py` — справочники: устройства, - города, источники трафика, карта сайта. Таблицы-литералы: доля живёт в - строке, которой принадлежит. -- `src/clickstream_generator/catalog.py` — каталог товаров из - `data/catalog/products.csv`, общего у генератора и словаря ClickHouse. -- `src/clickstream_generator/day.py` — день-функция: визиты, страницы, - атрибуция, устройство и гео. Там же правила резки визитов и шов, на - который садятся торговые события; прожитый день отдаёт обе половины — - поток событий и заказы. -- `src/clickstream_generator/commerce.py` — торговые события: что легло в - корзину, что из этого куплено, деньги, номера заказов и сырой `ecommerce`. - Своя случайность, поэтому правка торговли трафик не двигает. -- `src/clickstream_generator/orders.py` — заказы бэкенда: вторая проекция - покупки. Личность покупателя и деньги магазина — скидка по промокоду, - доставка, итог; своя случайность, ветвящаяся по дню рождения заказа. -- `src/clickstream_generator/ids.py` — номера событий: неповторяющиеся и - ниже 2^53. Обещание одно на обе половины дня, поэтому и живёт отдельно. -- `src/clickstream_generator/serialize.py` — канонический сериализатор: - единственное место, где событие целиком превращается в JSON. Порядок ключей, - все 47 колонок всегда и форма на проводе — ISO-8601. -- `src/clickstream_generator/sinks.py` — приёмники: файл (одно событие — одна - строка) и Kafka (одно событие — одно сообщение). Про содержимое они не - знают; там же довод, почему у сообщения нет ключа. -- `src/clickstream_generator/player.py` — проигрыватель: гонит дни в приёмник - пачкой или с темпом живого дня. Состояния не хранит. -- `src/clickstream_generator/cli.py` — интерфейс запуска: параметры - аргументами или окружением, логи в стандартный вывод, итог кодом возврата. -- `src/clickstream_generator/schema.py` — контракт схемы: чистые данные о - колонках выгрузки. Собственность генератора; из него выводятся сам - генератор, его валидация и описание выгрузки в доках. -- `src/clickstream_generator/schema_doc.py` — сборка «описания выгрузки» - ([`docs/formats/clickstream-event.md`](../docs/formats/clickstream-event.md)) - из контракта. Документ руками не правят — пересобирают. -- `src/clickstream_generator/inventory.py` — сборка паспорта мира - ([`data/world-inventory.json`](../data/world-inventory.json)): шапка паспорта - и хеши восьми дней, которыми наполняется стенд, плюс счётчики классов - расхождений по заказам для дней с закрытым окном. Руками не правят — - пересобирают целью `make inventory`. -- `tests/` — инварианты контракта, свежесть описания и обещания мира: - чистота от зерна, приток, гарантия двухкуковых пар, форма суточной волны - и сборка визитов по задокументированным правилам. Там же побайтовое - обещание, доведённое до диска: два прогона дня в файл дают тот же файл, а - строк в нём ровно столько, сколько событий. +Все перечисленные модули находятся в `src/clickstream_generator/`. + +| Модули | Назначение | +|---|---| +| `world.py`, `seeds.py`, `plan.py` | Настройки мира, случайные выборы и план посещений | +| `reference.py`, `catalog.py` | Устройства, города, источники трафика и каталог товаров | +| `day.py`, `commerce.py` | Визиты, события сайта и покупки | +| `orders.py` | Состояние заказов на дату слепка, изменения и задержки | +| `serialize.py`, `player.py`, `sinks.py` | Запись JSON и отправка в файл или Kafka | +| `cli.py` | Команды и параметры запуска | +| `schema.py`, `schema_doc.py` | Описания колонок события и сборка документа выгрузки | +| `inventory.py` | Сборка паспорта мира с контрольными суммами и счетчиками | + +Тесты в `tests/` проверяют формат, правила модели и повторяемость данных. +Описание выгрузки и [паспорт мира](../data/world-inventory.json) собираются +из кода; вручную их не редактируют. ## Команды -Проиграть день — раздел [«Как запустить -генератор»](../docs/architecture/world.md#как-запустить-генератор) -справочника мира и `--help`. +Для запуска из каталога `generator/` используйте +`uv run python -m clickstream_generator <команда>`. Справку можно открыть +через `--help`, в том числе после имени команды. -Из этого каталога — цели генератора живут здесь, а не в корне: +| Команда | Что отправляет | Особые параметры | +|---|---|---| +| `batch` | События дня без пауз | `--limit N` ограничивает число событий за весь запуск | +| `live` | События постепенно, с ускорением | `--speed X`, по умолчанию 60: сутки занимают около 24 минут | +| `snapshot` | Слепок заказов предыдущего дня | `--day D` задает день отправки | + +Для всех команд нужны номер дня `--day` и приемник: `--file` либо +`--brokers` вместе с `--topic`. Общие параметры `--seed` и `--days` +задают зерно и число последовательных дней. Параметры можно передать +и через окружение; имена и примеры собраны в +[справочнике запуска](../docs/architecture/world.md#как-запустить-генератор). +Прямой запуск генератора не продвигает сохраненную позицию времени стенда — +ей управляют [даги модельного времени](../docs/handbook/airflow-dags.md#даги-мира). + +Из каталога `generator/` доступны: - `make test` — тесты генератора; -- `make lint` — ruff: проверка и формат; -- `make typecheck` — ty: проверка типов; -- `make demo` — пройти стадии канонического D0 на одном экране; +- `make lint` — проверка кода и форматирования через ruff; +- `make typecheck` — проверка типов через ty; +- `make demo` — показать стадии расчета первого дня, D0; - `make docs` — пересобрать описание выгрузки; - `make inventory` — пересобрать паспорт мира. -В корне репозитория `make lint` тоже есть, но он про код стенда — даги и -Superset. Одноимённые цели за разными дверями охватывают разное: -[карта проверок](../docs/architecture/testing.md). - -Python и зависимости — через `uv`, версии закреплены в `uv.lock`: на этом -держится обещание побайтовой воспроизводимости (спека, раздел 2). +В корне `make lint` проверяет даги, код Superset и сборку карты архитектуры. +Корневые `make generate-batch` и `make generate-live` запускают события +через контейнер генератора. Состав проверок — в +[карте проверок](../docs/architecture/testing.md). +Python и зависимости запускаются через `uv`; версии закреплены в `uv.lock`. diff --git a/sql/ddl/20-ods-tables.sql b/sql/ddl/20-ods-tables.sql index 1e54f7d..d8415b7 100644 --- a/sql/ddl/20-ods-tables.sql +++ b/sql/ddl/20-ods-tables.sql @@ -1,49 +1,20 @@ --- ODS: типизированное событие и таблица ошибок разбора. --- --- Состав, имена и типы колонок списаны с «описания выгрузки» --- (docs/formats/clickstream-event.md) — так же, как в бою хранилище пишут по --- документации источника. Модуль контракта генератора здесь не читается: --- граница «трекер | хранилище» проходит по документу. --- Механика приёма и три класса брака — ADR 0005, конвенции имён, служебных --- колонок и сроков — docs/architecture/storage.md. - -- Локальная таблица события. +-- Поля следуют docs/formats/clickstream-event.md; purchaseRevenue хранится +-- в Decimal для точного счета денег. Правила приема — ADR 0005. +-- ReplacingMergeTree оставляет последнюю доставку по _load_ts среди строк +-- с одинаковым полным ORDER BY. Метка переносится из STG без пересчета. +-- Она не входит в ключ сортировки или партиции: иначе повторные доставки +-- не считались бы версиями одной строки. Для чтения служит ods.event_v с FINAL. -- --- Движок ReplacingMergeTree, колонка версии — _load_ts. Работа у версии ровно --- одна: схлопнуть повтор доставки. Метка не ставится здесь заново, а --- переносится из stg.hits_raw как есть — колонка отвечает на вопрос «когда --- строка приехала в хранилище», а не «когда её разобрали». У повтора --- содержимое то же самое, отличается только метка, поэтому какая из двух --- строк переживёт мерж, безразлично. +-- Партиция EventDate хранит модельный день события. CounterID на стенде +-- постоянен, EventDate постоянна внутри партиции; дальнейшая сортировка — +-- по intHash32(ClientID) и WatchID. WatchID отличает события посетителя: +-- без него они заменяли бы друг друга. -- --- Колонка версии не входит ни в ключ партиции, ни в ключ сортировки, и это --- не случайность: попади она туда — версии одной строки лягут в разные куски --- или в разные места ключа и не встретятся при мерже, то есть дедупликация --- перестанет работать молча. --- --- PARTITION BY EventDate — по дню события, а не по месяцу, как у Метрики. --- Отступление осознанное: дневная партиция здесь единица переобработки, и --- переиграть день X значит заменить одну партицию. Месячная партиция тянула --- бы за собой тридцать чужих дней. --- --- ORDER BY вырожден, и это учебный факт, а не недосмотр: CounterID на стенде --- константа (сайт один), EventDate константа внутри своей партиции — обе --- головные колонки ключа не различают ни одной строки. Реальная сортировка --- идёт по посетителю и событию: intHash32(ClientID), WatchID. Ключ написан в --- боевой форме «по сайту за период по посетителю» — на стенде она --- вырождается, в бою нет. Хвост WatchID работает ещё и на дедупликацию: без --- него ReplacingMergeTree схлопнул бы в одну строку все события посетителя за --- день. --- --- SAMPLE BY intHash32(ClientID) — по тому же выражению, что стоит в ключе --- сортировки (иначе семплирование не разрешено). Семплирование берёт целиком --- посетителей, а не события вразнобой, поэтому SAMPLE 0.1 не портит --- uniq-метрики. --- --- Sign всегда равен 1: это колонка формата, взятая без механики. В бою --- исправление записи шлют парой −1/+1 и считают через sum(Sign) поверх --- CollapsingMergeTree; наш генератор исправлений не шлёт, поэтому колонка --- есть, а механики за ней нет. +-- SAMPLE BY использует то же выражение, что ключ сортировки, и отбирает +-- посетителей целиком. Sign всегда равен 1: поле сохранено по формату +-- источника, но исправлений парой −1/+1 генератор не отправляет. + CREATE TABLE IF NOT EXISTS ods.event_rep ON CLUSTER clickstream_cluster ( WatchID UInt64, @@ -100,36 +71,24 @@ PARTITION BY EventDate ORDER BY (CounterID, EventDate, intHash32(ClientID), WatchID) SAMPLE BY intHash32(ClientID); --- Лицо слоя: пишем и читаем через него. Ключ шардирования — --- cityHash64(ClientID), а не сырой ClientID: структурированный числовой --- идентификатор перекашивает остаток по модулю числа шардов, хеш — нет. --- События одной куки при этом остаются на одном шарде, и сессионизация со --- склейкой идентичностей ниже по течению живут локально. --- --- У STG ключ другой (хеш строки), поэтому сырая строка и разобранное из неё --- событие почти всегда лежат на разных шардах. Пошардовые счётчики слоёв --- сходиться не должны — сверять слои можно только через _dist. +-- Запись и диагностика через Distributed; актуальные события — в ods.event_v. +-- cityHash64(ClientID) распределяет посетителей равномернее, чем остаток +-- самого структурированного идентификатора. События посетителя лежат вместе. +-- STG распределен по хешу raw, поэтому сообщение и разобранное событие могут +-- лежать на разных шардах. Общие счетчики сравнивают через _dist. + CREATE TABLE IF NOT EXISTS ods.event_dist ON CLUSTER clickstream_cluster AS ods.event_rep ENGINE = Distributed('clickstream_cluster', 'ods', 'event_rep', cityHash64(ClientID)); --- Локальная таблица ошибок разбора. +-- Ошибки разбора: исходный текст, класс и сведения о доставке. +-- У сообщения может не быть разобранного ClientID, поэтому шардирование — +-- по cityHash64(raw), партиция — по реальному дню загрузки. +-- ORDER BY группирует ошибки по классу и координатам Kafka. -- --- Сырой текст сообщения, метаданные доставки и класс брака. Разобранных полей --- у брака нет по определению: строка сюда попала как раз потому, что не --- разобралась. --- --- Движок — обычный ReplicatedMergeTree, без замены версий: у брака нет ключа --- сущности, схлопывать его не по чему. --- --- Ключи собственные. Шардирование — cityHash64(raw): ClientID у неразобранной --- строки взять неоткуда. ORDER BY (error_class, kafka_partition, --- kafka_offset): такую таблицу смотрят от класса брака, а внутри класса — по --- координатам доставки. Нарезка по дню загрузки, как у STG: модельного дня --- у брака тоже нет. --- --- Срок жизни — месяц, вдесятеро дольше сырых данных. Истеки брак вместе с ними — --- к моменту разбирательства не осталось бы ни того, ни другого. +-- Замены версий нет: надежного ключа события у брака может не быть. +-- TTL сохраняет ошибки месяц, дольше трехсуточного срока STG. + CREATE TABLE IF NOT EXISTS ods.event_errors_rep ON CLUSTER clickstream_cluster ( raw String, diff --git a/sql/ddl/30-ods-views.sql b/sql/ddl/30-ods-views.sql index 0e8a0ee..d7b81b2 100644 --- a/sql/ddl/30-ods-views.sql +++ b/sql/ddl/30-ods-views.sql @@ -1,8 +1,6 @@ --- ODS: каноническое чтение актуальных событий. --- --- Представление сохраняет поля и зерно ods.event_dist, но скрывает FINAL от --- потребителя. Поэтому физическая таблица остаётся местом диагностики --- доставленных версий, а точное чтение получает одно имя. +-- Актуальные события: FINAL выбирает последнюю доставку по _load_ts, +-- не дожидаясь фоновых слияний. Поля и типы остаются как в ods.event_dist. + CREATE VIEW IF NOT EXISTS ods.event_v ON CLUSTER clickstream_cluster AS SELECT * @@ -25,69 +23,24 @@ AS SELECT * FROM ods.order_dist FINAL; --- ODS: разбор STG в событие и в таблицу ошибок. +-- Разбор STG в события и ошибки. Два представления используют одинаковое +-- условие годности; второе берет его отрицание. Условие дает только 0 или 1: +-- NULL не выбрала бы ни одна ветвь. Функции разбора не вызывают исключений +-- на некорректном raw, чтобы одно сообщение не прервало прием. -- --- Матвью две, и вместе они обязаны делить поток без зазора и без нахлёста: --- строка, подошедшая обеим, задвоится, а не подошедшая ни одной — исчезнет --- молча. Держится это формой: годность считает один и тот же предикат из трёх --- частей, а вторая матвью берёт его буквальное отрицание. +-- Оба представления срабатывают на вставку в stg.hits_raw_dist, +-- до распределения по шардам. Порядок SELECT совпадает с целевой таблицей. -- --- Отсюда два запрета на выражения предиката, и оба серьёзные. +-- Прием проверяет объект JSON, точный набор ключей и разбор пяти полей: +-- WatchID, VisitID, ClientID, EventDate, UTCEventTime. Остальные поля +-- извлекаются по формату без отдельной проверки. Граница — ADR 0005. +-- arraySort с обеих сторон позволяет сравнить ключи независимо от порядка. -- --- Первый: ни одна его часть не возвращает NULL. Трёхзначная логика дала бы --- строку, которую не берёт ни «условие», ни «NOT условие», — и потерялась бы --- она бесшумно. Поэтому сравнения здесь дают 0 или 1, а разбор в Nullable --- заканчивается IS NOT NULL. --- --- Второй: ни одна его часть не бросает исключений. Упавшая матвью роняет --- вставку, офсеты не коммитятся, и потребление топика встаёт до починки — --- это единственное, на чём сейчас держится правило репозитория «грязные --- записи не валят пайплайн». Поэтому приведение Nullable к необнуляемому типу --- (assumeNotNull) стоит только в списке колонок, за предикатом, который NULL --- уже отсёк, а в самом предикате живут только функции семейства --- JSONExtract/*OrNull. Довод целиком — ADR 0005. --- --- Источник у обеих — stg.hits_raw_dist, лицо слоя, а не локальная таблица: --- потребитель цепляется к лицу, а матвью с источником-Distributed срабатывает --- на вставку именно в распределённую таблицу, до раскладки по шардам. --- --- Порядок колонок в SELECT совпадает с порядком в целевой таблице. - --- Годное событие: строка, прошедшая строгий приём. --- --- Строгий приём — это три вопроса: объект ли это JSON, тот ли набор ключей, --- разбираются ли в свой тип пять опорных колонок. Почему именно так — почему --- объект, а не валидность; почему сверка ключей заменяет сорок семь проверок --- на присутствие; почему опорных пять, а не сорок семь — ADR 0005, «Решение». --- --- arraySort ниже стоит с обеих сторон, и убирать его нельзя: без обёртки сорок --- семь CamelCase-имён пришлось бы держать ровно в байтовом порядке, а сбой --- порядка увёл бы в брак вообще всё, и молча (ADR 0005). --- --- Метка времени — единственная из пяти, кого разбирает не JSONExtract, и это --- измеренная необходимость, а не вкус. На проводе UTCEventTime уезжает в --- ISO-8601 с суффиксом зоны — «2026-06-01T12:34:56Z» (спека генератора, --- раздел 4), а JSONExtract с типом DateTime такую строку не берёт и отдаёт --- NULL. Оставь его здесь — и в брак уехали бы все события до единого. --- --- Формат назван буквально, а не отдан parseDateTimeBestEffort, и вот почему. --- Best-effort понимает десяток записей и на непонятной не краснеет, а --- достраивает недостающее: обрезанное «20:00:21» он превращает в первое --- января текущего года. Такая строка прошла бы строгий приём с тихо неверным --- временем — ровно с той порчей, ради которой класс key_field_unparsed и --- заведён. Источник у топика один и шлёт одну запись, так что широта здесь не --- нужна вовсе, а стоит она отключённой проверкой. Замеры — ADR 0005, --- «Что проверено». --- --- Третьим аргументом назван пояс — 'UTC'. Суффикс Z маска сверяет как букву и --- выбрасывает, зоны из строки не берёт вовсе, поэтому без имени функция читала --- бы показания часов по поясу сессии, а тот по умолчанию серверный. Тип --- колонки этого не чинит: разбор отдаёт готовое число, и колонка кладёт его --- как есть — пояс приёмника решал бы судьбу строки, а не числа. --- Правило и замер — docs/architecture/storage.md, «Часовые пояса». --- --- EventDate в такой подпорке не нуждается: дата уезжает как «2026-06-01», и --- JSONExtract её берёт. +-- UTCEventTime приходит строкой вида "2026-06-01T12:34:56Z". Ее разбирает +-- parseDateTimeOrNull с явной маской: JSONExtract в DateTime такую форму +-- не принимает, а BestEffort допускает неполные даты. Пояс UTC указан +-- отдельно, потому что Z в маске проверяется как буква. EventDate приходит +-- как "2026-06-01" и разбирается через JSONExtract. CREATE MATERIALIZED VIEW IF NOT EXISTS ods.event_mv ON CLUSTER clickstream_cluster TO ods.event_dist @@ -154,10 +107,8 @@ SELECT JSONExtract(raw, 'GoalsReached', 'Array(UInt32)') AS GoalsReached, JSONExtract(raw, 'ParsedParamsKey1', 'Array(String)') AS ParsedParamsKey1, JSONExtract(raw, 'purchaseID', 'Array(String)') AS purchaseID, - -- Выгрузка несет JSON-числа типа Float64, но деньги хранятся в Decimal. - -- JSONExtract разбирает их сразу, без промежуточного двоичного числа: - -- синтаксис подтвержден документацией ClickHouse через Context7 и - -- запросом к стендовому ClickHouse 26.3 при исполнении #155. + -- purchaseRevenue приходит числом рублей и разбирается сразу в Decimal. + -- Прямой разбор сверяли через Context7 и на ClickHouse 26.3 (#155). JSONExtract(raw, 'purchaseRevenue', 'Array(Decimal(18, 2))') AS purchaseRevenue, JSONExtract(raw, 'purchaseCurrency', 'Array(String)') AS purchaseCurrency, JSONExtract(raw, 'purchaseCoupon', 'Array(String)') AS purchaseCoupon, @@ -168,25 +119,16 @@ SELECT JSONExtract(raw, 'productQuantity', 'Array(UInt64)') AS productQuantity, JSONExtract(raw, 'productEventType', 'Array(String)') AS productEventType, JSONExtract(raw, 'ecommerce', 'String') AS ecommerce, - -- Метка загрузки переносится из STG как есть, а не ставится заново: - -- она отвечает на вопрос «когда строка приехала в хранилище». Поставь - -- здесь now64(3) — и колонка версии молча ответила бы на другой вопрос. + -- _load_ts остается временем исходной доставки и версией события. _load_ts FROM stg.hits_raw_dist WHERE is_object AND keys_match AND key_fields_parsed; --- Брак: буквальное отрицание того же предиката. --- --- Классы брака пересекаются, поэтому проверяются по порядку, а в error_class --- пишется первый совпавший. Скаляр проваливает и проверку на объект, и сверку --- ключей — JSONExtractKeys от него даёт пустой массив (измерено на стенде --- 7 августа 2026 года), — и без объявленного порядка попал бы то в один --- класс, то в другой. --- --- Предикат повторён здесь дословно, и это выбор, а не безвыходность: назвать --- его один раз на две матвью позволил бы CREATE FUNCTION. Отвергнуто — условие --- разбора ушло бы за имя, в отдельный объект со своей жизнью, и слой перестал --- бы читаться по своему же DDL. +-- Ошибки: отрицание того же условия годности. +-- Класс определяется первым нарушенным условием. Например, JSON-строка +-- не проходит и проверку объекта, и проверку ключей, но получает not_an_object. +-- Условие повторено рядом с запросом, без отдельной функции ClickHouse. + CREATE MATERIALIZED VIEW IF NOT EXISTS ods.event_errors_mv ON CLUSTER clickstream_cluster TO ods.event_errors_dist AS diff --git a/sql/ddl/40-stg-views.sql b/sql/ddl/40-stg-views.sql index 4a20282..a0e2d8b 100644 --- a/sql/ddl/40-stg-views.sql +++ b/sql/ddl/40-stg-views.sql @@ -1,26 +1,16 @@ --- STG: матвью приёма — из чтеца в таблицу STG. +-- Прием событий из Kafka в STG. +-- Подключение этого представления запускает чтение топика. Поэтому файл 40 +-- применяется после таблиц и представлений разбора из файлов 20 и 30: +-- иначе часть сообщений сохранится в STG, но не попадет в ODS. -- --- Номер 40, а не 11, и пропуск в нумерации намеренный. Файлы 20-ods-tables.sql --- и 30-ods-views.sql приносит #43, и матвью приёма обязана создаваться после --- матвью разбора: Kafka-движок начинает читать топик ровно тогда, когда к нему --- привязывают первую матвью. Создай её раньше разбора — и всё, что доедет в --- зазоре, ляжет в STG и не попадёт в ODS никуда, ни в событие, ни в ошибки. --- На пустом топике зазор безвреден, поэтому первый прогон о нём не скажет: --- проснётся он, когда тома ClickHouse снесены, а данные Kafka целы, то есть на --- обычной отладке. Нумерацию здесь не «приводить в порядок». +-- Запись через stg.hits_raw_dist распределяет сообщения по ключу таблицы. +-- Она фоновая; возможную потерю при сбое описывает раздел «Приём» +-- в docs/architecture/storage.md. -- --- Пишем в stg.hits_raw_dist, а не в локальную таблицу: раскладку по шардам --- обязан определять ключ шардирования, а не то, какая нода случайно читала --- топик. Вставка при этом фоновая — окно потери принято осознанно, довод --- целиком в docs/architecture/storage.md, раздел «Приём». --- --- Служебные колонки заполняются выражением здесь, а не DEFAULT в таблице. --- Для consumer_host это обязательно: проверено на стенде 5 августа 2026 года — --- DEFAULT hostName() вычисляется на шарде-получателе и назвал бы не ту ноду, --- которая читала топик. hostName() в SELECT снимается на вставляющей ноде, --- то есть отвечает ровно на нужный вопрос. --- --- Порядок колонок в SELECT совпадает с порядком в целевой таблице. +-- Сведения о доставке вычисляются здесь, на ноде чтения Kafka. +-- DEFAULT hostName() только в локальной таблице назвал бы шард-получатель. +-- Порядок SELECT совпадает с порядком колонок цели. + CREATE MATERIALIZED VIEW IF NOT EXISTS stg.hits_raw_mv ON CLUSTER clickstream_cluster TO stg.hits_raw_dist AS diff --git a/sql/dds/order_rebuild.sql b/sql/dds/order_rebuild.sql index a43a053..1479e21 100644 --- a/sql/dds/order_rebuild.sql +++ b/sql/dds/order_rebuild.sql @@ -1,21 +1,9 @@ --- DDS: сборка дней заказа в таблицу-двойник перед заменой партиций. +-- Сборка выбранных дней в донор перед заменой партиций DDS. +-- Донор очищается в начале: иначе новые строки добавились бы к прежним. +-- После сборки он остается для диагностики до следующего запуска. -- --- Двойник очищается целиком в начале сборки, а не в конце: собранное остаётся --- лежать до следующего запуска, и разбор происшествия видит ровно то, чем --- подменили день. Очистка обязана быть первой ещё и по делу — вставка поверх --- прежних строк удвоила бы день в доноре, а REPLACE PARTITION перенёс бы --- удвоение в цель. --- --- Дни собираются одной вставкой, а не по дню за раз: источник читается один --- раз на весь отрезок. Кто раскладывает строки по шардам — вставка через --- Distributed, ключ cityHash64(order_id) — разобрано в DDL двойника. --- --- Чистит двойник DROP PARTITION ALL, а не TRUNCATE: последнее — отдельное --- право, роли etl оно не выдано, а снятие всех партиций --- входит в ALTER, который у неё есть (ADR 0007). Замерено: TRUNCATE отвечает --- ролью etl кодом 497, DROP PARTITION ALL проходит. Расширять права ради --- операции, для которой уже есть разрешённая форма, значило бы платить --- доступом за синоним. +-- DROP PARTITION ALL доступен роли etl через ALTER; отдельного права +-- TRUNCATE у нее нет (ADR 0007). Весь отрезок дней собирается одной вставкой. ALTER TABLE dds.order_stage_rep ON CLUSTER clickstream_cluster DROP PARTITION ALL; @@ -37,18 +25,13 @@ INSERT INTO dds.order_stage_dist _load_id, _load_ts ) --- Позиции разбираются здесь один раз, на границе ODS → DDS. Цена на проводе — --- строка с двумя знаками, и JSONExtract читает её прямо в Decimal: замерено на --- живом слепке, промежуточный toDecimal64 не нужен. --- --- Разбор по типу молчалив, и знать об этом надо заранее. Строгий приём ODS --- проверил только, что items — массив, внутрь он не смотрел; здесь тоже никто --- не смотрит. Замерено: qty вида "oops" даёт 0, пропущенная цена даёт 0, --- скаляр вместо объекта выпадает из массива целиком. На каноническом слепке --- генератора такого не бывает, и сумма «цена × количество» сходится с --- items_total у всех заказов, но сходится она по свойству источника, а не --- потому, что её здесь кто-то проверяет. Постоянное место такой проверки — --- контур DQ этапа 4; пока она делается руками при приёмке. +-- Разбор позиций из JSON: цена в строке сразу превращается в Decimal. +-- ODS проверил только тип массива items. Неверное содержимое здесь может +-- дать значения по умолчанию: qty вида "oops" и пропущенная цена дают 0, +-- скаляр вместо объекта выпадает из массива. Наблюдения — +-- docs/architecture/storage.md, «Что проверено». +-- Проверку суммы позиций участник добавляет в docs/labs/5-1-order-checks.md. + WITH JSONExtract( items, 'Array(Tuple(sku String, qty UInt32, price Decimal(18, 2)))' ) AS lines @@ -56,8 +39,7 @@ SELECT order_id, user_id, status, - -- День заказа и оба времени переезжают в пояс счётчика: DDS обслуживает - -- человека с дашбордом, и пояс называется здесь в последний раз. + -- День и время в Europe/Samara: витрины используют готовые колонки. toDate(toTimeZone(created_at, 'Europe/Samara')), toTimeZone(created_at, 'Europe/Samara'), toTimeZone(updated_at, 'Europe/Samara'), @@ -68,22 +50,17 @@ SELECT arrayMap(line -> line.sku, lines), arrayMap(line -> line.qty, lines), arrayMap(line -> line.price, lines), - -- Координаты запуска, собравшего день: у всех строк дня они одни. + -- Метки этой сборки DDS, общие для всех вставляемых строк. {load_id:String}, now64(3, 'UTC') FROM ods.order_v WHERE toDate(toTimeZone(created_at, 'Europe/Samara')) BETWEEN {first_day:Date} AND {last_day:Date} - -- Вторая граница не меняет отбор, а открывает источнику отсечение - -- партиций: ODS нарезан по toDate(created_at) в UTC, и условие на - -- вычисленный день заказа для этого ключа непрозрачно — сервер прочёл бы - -- все партиции. Сутки пояса счётчика ложатся ровно на две даты UTC, - -- поэтому день слева берётся с запасом. + -- Дополнительный фильтр сокращает чтение партиций ODS по UTC. + -- Модельные сутки Europe/Samara пересекают две даты UTC, поэтому + -- левая граница берется на день раньше. Точный отбор — условием выше. AND toDate(created_at) BETWEEN {first_day:Date} - 1 AND {last_day:Date} --- Тот же ноль, что в приёме заказов: заводская двойка исполнила бы вставку --- локально на каждом шарде, минуя ключ шардирования цели, и строки унаследовали --- бы раскладку источника. Ключи ODS и DDS здесь совпадают, и раскладка вышла бы --- верной — но по совпадению, а не по контракту (docs/architecture/storage.md, --- «Раскладка по шардам», капкан #137). +-- Ждем доставки на шарды и распределяем строки по ключу цели +-- (docs/architecture/storage.md, «Раскладка по шардам»). SETTINGS distributed_foreground_insert = 1, parallel_distributed_insert_select = 0; diff --git a/sql/dds/order_replace.sql b/sql/dds/order_replace.sql index 3af9d70..d2b7d6a 100644 --- a/sql/dds/order_replace.sql +++ b/sql/dds/order_replace.sql @@ -1,28 +1,16 @@ --- DDS: подмена одного дня заказов собранной партицией двойника. +-- Замена одного дня DDS готовой партицией донора. +-- Повтор при неизменном источнике сохраняет деловые данные без дублей. -- --- Это и есть идемпотентность слоя. Повтор прогона безопасен, а «дыхание» --- статусов внутри окна выражается сменой всей партиции, а не правкой строк. +-- ON CLUSTER выполняет замену на обеих нодах по локальным таблицам. +-- Донор и цель имеют одинаковую структуру и ключ шардирования +-- (sql/ddl/45-dds-tables.sql); REPLACE PARTITION строки не перераспределяет. +-- Замена атомарна внутри шарда. Между заменами на разных шардах читатель +-- может увидеть часть прежних и часть новых данных. После сбоя этот период +-- длится до успешного повтора. Пустая партиция донора удаляет день из цели. -- --- Замена идёт по локальным таблицам, шард за шардом: ON CLUSTER рассылает её --- на обе ноды, а по Distributed то же самое не работает — операции с --- партициями отвечают кодом 48 (docs/architecture/storage.md, «Что --- проверено»). Правильность пошардовой замены держится на том, что двойник --- шардирован тем же ключом, что цель, — см. sql/ddl/45-dds-tables.sql. --- --- Отсюда граница обещания: атомарность здесь пошардовая. Внутри шарда подмена --- мгновенна — читатель видит либо прежний день целиком, либо новый; но две --- локальные замены независимы, и в зазоре между ними чтение через Distributed --- соберёт ответ из нового дня одного шарда и прежнего дня другого. Зазор --- короткий, а отказ одной из двух замен растягивает его до повторного прогона. --- --- Пустая партиция двойника — законный случай, а не сбой: день, в котором --- заказов нет, снимает день из цели, и слой перестаёт показывать то, чего в --- источнике больше нет. --- --- День подставляет Jinja, а не параметр ClickHouse: в ALTER … PARTITION --- сервер параметров не подставляет — замерено, отвечает синтаксической --- ошибкой. Общее правило репозитория «параметры передаются отдельно, а не --- строкой» здесь не нарушено по существу: значение пришло из ответа самого --- ClickHouse на запрос объёма, а не снаружи стенда. +-- День подставляет Jinja: параметры ClickHouse в ALTER ... PARTITION +-- не поддерживаются в закрепленной версии 26.3. Значение получено +-- из результата order_scope.sql; замер — docs/architecture/storage.md. + ALTER TABLE dds.order_rep ON CLUSTER clickstream_cluster REPLACE PARTITION '{{ params.day }}' FROM dds.order_stage_rep; diff --git a/sql/dds/order_scope.sql b/sql/dds/order_scope.sql index 3b5c8b0..e83a1f3 100644 --- a/sql/dds/order_scope.sql +++ b/sql/dds/order_scope.sql @@ -1,37 +1,17 @@ --- DDS: объём пересборки заказов — с какого дня по какой собирает этот запуск. +-- Дни пересборки заказов: first_day и last_day, обе границы включены. +-- Текущее окно — дни оформления заказов из последнего snapshot_date. +-- Если в DDS нет дня, который есть в ODS, отрезок расширяется до первого +-- такого дня. Это восстанавливает пропуски даже за пределами текущего окна. -- --- Параметров у дага нет: объём он выводит сам из состояния слоёв --- (docs/architecture/etl.md, «Объём прогона»). Ответ — две даты, границы --- непрерывного отрезка; список дней между ними разворачивает даг. Пустой --- источник даёт пустой ответ, и даг на нём останавливается с внятной ошибкой --- вместо зелёного прогона, собравшего ничего. +-- Дата слепка у актуальной версии показывает последнее наблюдение заказа +-- (ADR 0014). Поэтому окно берется из данных, без копирования константы +-- генератора. FINAL здесь не нужен: created_at одинаков у всех версий, +-- а строки последнего слепка сохраняются при слияниях. -- --- Файл кончается без точки с запятой: даг читает его целиком и шлёт одним --- запросом, дописывая формат ответа, — точка с запятой сделала бы из этого --- мультизапрос, который ClickHouse не принимает. --- --- Объём складывается из двух частей. Первая — дышащее окно: дни, которые нёс --- последний слепок, то есть те, чьё состояние источник ещё может изменить. --- Вторая — хвост от первого дня, которого в слое нет: первая сборка застаёт --- слой пустым и собирает весь стартовый мир, а после разгона мира пачкой слой --- догоняет уехавший вперёд источник. --- --- **Хвост считается от первого недостающего дня, а не от последнего --- собранного.** Разница не косметическая: замены партиций идут отдельными --- задачами и падают порознь, поэтому упавший день оставляет дыру, пока --- соседние дни собираются дальше. Считай мы прогресс по максимуму собранного --- дня, дыра позади окна не заросла бы никогда, а прогоны продолжали бы --- зеленеть над щербатым слоем. Цена честного ответа — проход по дню заказа --- в ODS. --- --- Окно читается из ярлыка слепка, а не вычисляется вычитанием окна --- изменяемости: у выжившей строки snapshot_date — последний слепок, который --- нёс этот заказ (ADR 0014). Поэтому оба края окна знает сам источник, а --- копии его константы на стороне стенда больше нет. --- --- FINAL не нужен ни здесь, ни ниже: время рождения заказа у всех его версий --- одно, а строки последнего слепка несут максимум колонки версии — мерж --- оставляет как раз их. +-- Пустой ODS не дает строки результата; даг завершится ошибкой. +-- Точка с запятой в конце не нужна: клиент дописывает FORMAT ответа. +-- Порядок пересборки — docs/architecture/etl.md, «Объём прогона». + WITH (SELECT max(snapshot_date) FROM ods.order_dist) AS last_snapshot, ( @@ -44,10 +24,9 @@ WITH FROM ods.order_dist WHERE snapshot_date = last_snapshot ) AS window_end, - -- GLOBAL обязателен: обе таблицы распределённые, и обычный NOT IN сверял - -- бы дни источника с локальной частью слоя на каждом шарде — день, чьи - -- заказы легли на соседний шард, показался бы недостающим. - -- minOrNull отличает «дыр нет» от даты по умолчанию. + -- GLOBAL NOT IN сравнивает дни со всем DDS. Локальное сравнение могло бы + -- счесть день пропущенным, если его заказы находятся на соседнем шарде. + -- minOrNull возвращает NULL, когда пропущенных дней нет. ( SELECT minOrNull(day) FROM ( @@ -57,7 +36,7 @@ WITH WHERE day GLOBAL NOT IN (SELECT order_date FROM dds.order_dist) ) AS first_missing_day SELECT - -- Дыр нет — начинаем с окна; есть — с первой из них, если она старше окна. + -- Начало окна или первый пропущенный день, если он раньше. least(window_start, coalesce(first_missing_day, window_start)) AS first_day, window_end AS last_day WHERE (SELECT count() FROM ods.order_dist) > 0 diff --git a/sql/ods/order_parse.sql b/sql/ods/order_parse.sql index a5a23b5..8bfa72a 100644 --- a/sql/ods/order_parse.sql +++ b/sql/ods/order_parse.sql @@ -1,24 +1,15 @@ --- Разбор одного среза заказов: сначала годные версии, затем брак. +-- Разбор сохраненной порции заказов: сначала годные версии, затем ошибки. +-- Макрос Jinja подставляет одно условие годности в обе вставки. +-- Вторая вставка берет его отрицание, чтобы каждая строка получила результат. +-- Условие должно давать только 0 или 1 и не вызывать исключений на неверном raw. -- --- Обе вставки должны делить один предикат. Локальный макрос Jinja --- разворачивает его в каждое выражение до того, как оператор разделит файл по --- точкам с запятой. Так контракт остаётся рядом с запросами и не становится --- третьим SQL-файлом, который нельзя исполнить самостоятельно. --- --- Строгий приём проверяет форму провода и ничего сверх неё. Ключей одиннадцать, --- и проверяются все: десять скалярных значений прямо образуют типизированную --- строку заказа. Внутрь items приём не смотрит — содержимое позиций, переходы --- статуса и равенства сумм остаются ниже этой границы --- (docs/architecture/orders/ingestion.md, «Граница строгого приёма»). --- --- У общего предиката два обязательных свойства. Он всегда даёт 0 или 1, не --- NULL: иначе строку не возьмут ни условие, ни его отрицание. И он не бросает --- исключений на произвольном raw: одна грязная строка не должна ронять весь --- переход, ради этого и существует таблица брака. +-- Проверяются одиннадцать полей верхнего уровня. Внутри items поля не проверяются; +-- позиции разбирает DDS, а равенства сумм проверяются отдельно. +-- Точные правила — docs/architecture/orders/ingestion.md, «Граница строгого приёма». {% macro order_wire_contract() %} WITH - -- Каноническое время провода: UTC, ровно три знака долей секунды. + -- Время источника: UTC, ровно три знака долей секунды. 'yyyy-MM-dd\'T\'HH:mm:ss.SSS\'Z\'' AS ts_mask, JSONType(raw) = 'Object' AS is_object, arraySort(JSONExtractKeys(raw)) = arraySort([ @@ -28,15 +19,14 @@ WITH ]) AS keys_match, JSONType(raw, 'order_id') = 'String' AND JSONType(raw, 'status') = 'String' - -- Целое у JSONType зовётся двумя именами, и нужны оба: с одним - -- Int64 законный идентификатор за 2^63 уехал бы в брак. + -- JSONType различает Int64 и UInt64; принимаются оба, если значение + -- разбирается в UInt64, в том числе идентификаторы больше 2^63−1. AND JSONType(raw, 'user_id') IN ('Int64', 'UInt64') AND JSONExtract(raw, 'user_id', 'Nullable(UInt64)') IS NOT NULL AND JSONType(raw, 'items') = 'Array' - -- Форму держит регулярка, реальность — разбор, и порознь они дырявы: - -- маска берёт «2026-6-3T…» без ведущих нулей, а регулярка пропускает - -- 30 февраля. Обе половины измерены — docs/architecture/storage.md, - -- «Что проверено». + -- Регулярное выражение проверяет точную форму, включая ведущие нули. + -- Разбор проверяет само время: например, отвергает 30 февраля. + -- Замеры функций — docs/architecture/storage.md, «Что проверено». AND arrayAll(k -> JSONType(raw, k) = 'String' AND match(JSONExtractString(raw, k), @@ -44,9 +34,8 @@ WITH AND parseDateTime64InJodaSyntaxOrNull( JSONExtractString(raw, k), ts_mask, 'UTC') IS NOT NULL, ['created_at', 'updated_at']) - -- Деньги — строка с ровно двумя знаками после точки. Регулярка держит - -- форму, разбор — вместимость: у строки из двадцати девяток форма - -- хороша, а в Decimal(18, 2) она не влезает. + -- Сумма в рублях: строка с двумя знаками после точки, без минуса. + -- После проверки формы проверяется возможность разбора в Decimal. AND arrayAll(k -> JSONType(raw, k) = 'String' AND match(JSONExtractString(raw, k), '^\\d+\\.\\d{2}$') @@ -60,11 +49,9 @@ WITH is_object AND keys_match AND fields_valid AS row_is_valid {% endmacro %} --- Годные версии заказа. Срез читается по _load_id — тому же, что проставил --- забор: разбор идёт по неизменной порции, а не по «всему, что появилось». --- --- Разобранные значения берёт assumeNotNull: обнуляемый разбор стоит за --- предикатом, который NULL уже отсёк, и приведение здесь не может упасть. +-- Годные строки сохраненного среза по _load_id запуска приема. +-- WHERE исключает строки с NULL в разобранных полях; assumeNotNull +-- приводит оставшиеся значения к типам целевой таблицы. INSERT INTO ods.order_dist {{ order_wire_contract() }} @@ -81,30 +68,25 @@ SELECT assumeNotNull(toDecimal64OrNull(JSONExtractString(raw, 'discount'), 2)) AS discount, assumeNotNull(toDecimal64OrNull(JSONExtractString(raw, 'delivery'), 2)) AS delivery, assumeNotNull(toDecimal64OrNull(JSONExtractString(raw, 'total'), 2)) AS total, - -- items кладётся сырым фрагментом JSON, а не разобранной структурой. + -- items сохраняется строкой JSON для разбора позиций в DDS. JSONExtractRaw(raw, 'items') AS items, toDate(assumeNotNull(parseDateTimeInJodaSyntaxOrNull( JSONExtractString(raw, 'snapshot_date'), 'yyyy-MM-dd', 'UTC'))) AS snapshot_date, - -- Метки запуска и прибытия переносятся как есть. Поставь здесь now64(3) — - -- и _load_ts молча ответила бы на другой вопрос: «когда разобрали». + -- _load_ts остается временем приема в STG; _load_id связывает с исходной порцией. _load_id, _load_ts FROM stg.orders_raw_dist WHERE _load_id = {load_id:String} AND row_is_valid --- Заводская двойка parallel_distributed_insert_select исполняет вставку из --- Distributed в Distributed локально на каждом шарде, минуя ключ шардирования --- цели, — так заказы и расползлись по обоим шардам. Ноль отдаёт раскладку --- инициатору, по ключу цели. Капкан целиком — docs/architecture/storage.md, --- «Раскладка по шардам». +-- Ноль заставляет распределить результат по ключу цели. При значении 2 +-- вставка может сохранить раскладку STG, где ключ — raw, вместо order_id. +-- Подробнее — docs/architecture/storage.md, «Раскладка по шардам». SETTINGS distributed_foreground_insert = 1, parallel_distributed_insert_select = 0; --- Брак: тот же срез и буквальное отрицание того же предиката. --- --- Классы перекрываются, поэтому проверяются по порядку, а в error_class идёт --- первый совпавший: скаляр проваливает и проверку на объект, и сверку ключей — --- без объявленного порядка он попал бы то в один класс, то в другой. +-- Ошибки: тот же срез и отрицание условия годности. +-- Класс определяется первым нарушенным условием: например, JSON-строка +-- не проходит и проверку объекта, и проверку ключей, но получает not_an_object. INSERT INTO ods.order_errors_dist {{ order_wire_contract() }} @@ -124,7 +106,6 @@ SELECT _load_ts FROM stg.orders_raw_dist WHERE _load_id = {load_id:String} AND NOT row_is_valid --- Тот же ноль: ключ брака сегодня совпадает с ключом STG, и локальная --- запись легла бы верно — но по совпадению, а не по контракту. +-- Та же настройка: распределяем строки по ключу целевой таблицы ошибок. SETTINGS distributed_foreground_insert = 1, parallel_distributed_insert_select = 0; diff --git a/sql/stg/orders_raw_load.sql b/sql/stg/orders_raw_load.sql index b6a2223..642d903 100644 --- a/sql/stg/orders_raw_load.sql +++ b/sql/stg/orders_raw_load.sql @@ -1,15 +1,11 @@ --- Забор порции слепка заказов из Kafka в STG. +-- Порция сообщений заказов из Kafka в STG, без разбора JSON. +-- Один SELECT читает одну порцию; непрочитанное остается до следующего запуска. +-- Полнота слепка не проверяется. Если после чтения вставка упадет, подтвержденные +-- сообщения уже не перечитаются (docs/architecture/orders/ingestion.md). -- --- Забор — один прямой SELECT, без цикла до пустоты: одна порция ClickHouse --- берёт десятки тысяч сообщений, а слепок дня — порядка полутора тысяч строк. --- Короткая порция оставит хвост до следующего прогона, а отказ после чтения --- унесёт прочитанное с собой: офсеты коммитятся в момент чтения. Граница --- целиком — ADR 0008, «Следствия». --- --- stream_like_engine_allow_direct_select разрешает читать чтеца запросом; вторая --- половина пары объявлена на самой таблице (sql/ddl/10-stg-tables.sql). --- distributed_foreground_insert = 1 — конвенция ETL-вставок стенда: задача не --- должна зеленеть раньше, чем строки легли на шарды. +-- stream_like_engine_allow_direct_select разрешает прямое чтение Kafka; +-- kafka_commit_on_select включен в sql/ddl/10-stg-tables.sql. +-- distributed_foreground_insert = 1 ждет записи на шарды до завершения задачи. INSERT INTO stg.orders_raw_dist SELECT