diff --git a/dags/orders_ingest.py b/dags/orders_ingest.py index e655077..566c68b 100644 --- a/dags/orders_ingest.py +++ b/dags/orders_ingest.py @@ -101,8 +101,8 @@ def orders_ingest(): logging.info("срез разобран: _load_id %s", load_id) pull_batch("stg/orders_raw_load.sql") >> parse_batch( - "ods/order_snapshot_load.sql", - "ods/order_snapshot_errors_load.sql", + "ods/order_load.sql", + "ods/order_errors_load.sql", ) diff --git a/docs/adr/0010-order-versions-in-ods.md b/docs/adr/0010-order-versions-in-ods.md index 86d6b70..8d5fca2 100644 --- a/docs/adr/0010-order-versions-in-ods.md +++ b/docs/adr/0010-order-versions-in-ods.md @@ -9,7 +9,7 @@ команде Airflow, но порция чтения больше не считается полным слепком и не публикуется заменой партиции `snapshot_date`. Прочитанные строки получают `_load_id` запуска, разбираются из одного среза STG в годные строки и ошибки, -а `ods.order_snapshot` хранит принятые версии заказа в +а `ods.order` хранит принятые версии заказа в `ReplacingMergeTree(updated_at)`. Ключ сущности — `order_id`, партиция строится от неизменного `created_at`, все версии ключа направляются на один шард. diff --git a/docs/architecture/orders/README.md b/docs/architecture/orders/README.md index 53af499..693851a 100644 --- a/docs/architecture/orders/README.md +++ b/docs/architecture/orders/README.md @@ -37,7 +37,7 @@ `person_id` заказ показывает как `user_id`, кликстрим остаётся анонимным — [мост к склейке](identity.md). - **Приём — пакетный забор**: одно прямое чтение Kafka в STG, два - `INSERT SELECT` в типизированный ODS и таблицу ошибок; `ods.order_snapshot` + `INSERT SELECT` в типизированный ODS и таблицу ошибок; `ods.order` принимает версии заказа на `ReplacingMergeTree(updated_at)` — [приём из Kafka в ODS](ingestion.md). - **Стартовый мир отправляет семь слепков** (дни 0…6): у последнего прожитого diff --git a/docs/architecture/orders/fate.md b/docs/architecture/orders/fate.md index 4a99c94..117aee7 100644 --- a/docs/architecture/orders/fate.md +++ b/docs/architecture/orders/fate.md @@ -61,7 +61,7 @@ **Что из двух дорог видно.** В `dds.order` дороги неразличимы — там последняя версия; различает их история версий: сырьё STG и физические версии -`ods.order_snapshot` до фоновых слияний, а в витринах — выручка дня, которая +`ods.order` до фоновых слияний, а в витринах — выручка дня, которая сначала выросла, потом убыла. Полное различение не обещано: слепок — состояние на границе суток, и оплата с отменой в один день в сырье неразличимы; то же у сильно опоздавших, приехавших уже терминальными. Точную форму этого урока diff --git a/docs/architecture/orders/ingestion.md b/docs/architecture/orders/ingestion.md index d67efcc..02a5220 100644 --- a/docs/architecture/orders/ingestion.md +++ b/docs/architecture/orders/ingestion.md @@ -43,12 +43,12 @@ Один следующий `task_id` отвечает за весь переход STG → ODS. Внутри него два последовательных `INSERT SELECT` читают неизменный срез по `_load_id`: первый -пишет годные строки в `ods.order_snapshot`, второй — брак в -`ods.order_snapshot_errors`. Транзакции между запросами нет. При частичном сбое -Airflow повторяет весь `task_id`; одинаковые исходные строки и служебные метки -не вычисляются заново. Запросы лежат рядом с целями: -[`order_snapshot_load.sql`](../../../sql/ods/order_snapshot_load.sql) и -[`order_snapshot_errors_load.sql`](../../../sql/ods/order_snapshot_errors_load.sql). +пишет годные строки в `ods.order`, второй — брак в `ods.order_errors`. +Транзакции между запросами нет. При частичном сбое Airflow повторяет весь +`task_id`; одинаковые исходные строки и служебные метки не вычисляются заново. +Запросы лежат рядом с целями: +[`order_load.sql`](../../../sql/ods/order_load.sql) и +[`order_errors_load.sql`](../../../sql/ods/order_errors_load.sql). Условия запросов взаимодополняющие: один общий предикат определяет брак, а годная ветвь использует его буквальное отрицание. Все функции предиката @@ -90,9 +90,9 @@ JSON-объектом с точным набором ключей: `order_id`, ` ## Роль ODS -`ods.order_snapshot_rep` хранит физически принятые версии в +`ods.order_rep` хранит физически принятые версии в `ReplacingMergeTree(updated_at)`. Ключ сортировки — `order_id`, партиция — день -неизменного `created_at`. `ods.order_snapshot_dist` шардирует по +неизменного `created_at`. `ods.order_dist` шардирует по `cityHash64(order_id)`: только так все версии заказа попадают на один шард и `FINAL` даёт корректный результат через распределённую таблицу. @@ -174,7 +174,7 @@ JSON-объектом с точным набором ключей: `order_id`, ` После исправления [#102](https://git.dementev.space/ddmitry/clickstream-data-platform/issues/102) честный прогон дня снят на живом стенде 20 августа 2026 года: все 1694 строки -слепка дня 7 приняты, `ods.order_snapshot_errors` осталась пустой. У заказа +слепка дня 7 приняты, `ods.order_errors` осталась пустой. У заказа `20260608-0007`, который прежде приходил с отрицательным итогом, теперь `5490.00 − 823.50 + 299.00 = 4965.50`. diff --git a/docs/architecture/orders/inventory.md b/docs/architecture/orders/inventory.md index e790ee5..5ddba60 100644 --- a/docs/architecture/orders/inventory.md +++ b/docs/architecture/orders/inventory.md @@ -25,6 +25,6 @@ пользователей и число двухкуковых пар растут, пока мир едет, — опоры из них не выходит; генератор сторожит хеш, транспорт — счёт событий. - **Опись описывает мир, а не доставку.** Работа генератора кончается на - Kafka: дошли ли байты до `ods.order_snapshot` — вопрос стенда и его + Kafka: дошли ли байты до `ods.order` — вопрос стенда и его проверок. Поэтому счёта строк у слепка в описи нет; понадобится проверка приёма заказов — число заведётся вместе с ней. diff --git a/docs/architecture/storage.md b/docs/architecture/storage.md index 2dad3a2..5702a25 100644 --- a/docs/architecture/storage.md +++ b/docs/architecture/storage.md @@ -60,7 +60,7 @@ CSV-файла. Живёт он вне цепочки слоёв, в зоне `d приходит от транспорта: слой хранит то, что доехало по топику, и зовётся именем топика — `hits_raw` от `hits`, `orders_raw` от `orders`. В типизированных слоях имя приходит от предметной области и стоит в единственном числе: `event`, -`order_snapshot`, `session`. Граница между «как привезли» и «что это такое» +`order`, `session`. Граница между «как привезли» и «что это такое» проходит по STG, и имена её показывают. ## Раскладка по шардам @@ -91,7 +91,7 @@ CSV-файла. Живёт он вне цепочки слоёв, в зоне `d по `cityHash64(order_id)`; ключи остальных частей будущей модели и витрин появятся вместе с ними. -Замена партиций не используется для `ods.order_snapshot`: версии прошлых дней +Замена партиций не используется для `ods.order`: версии прошлых дней доливаются, а прямое чтение Kafka не задаёт границы полного слепка ([ADR 0010](../adr/0010-order-versions-in-ods.md)). Если партиционная пересборка понадобится будущим объектам DDS или DM, партиция-донор должна быть заранее @@ -141,7 +141,7 @@ kafka_offset)`: разбор полётов идёт от «какое сооб заказов — пакетным шагом. Дальше метка переносится в ODS как есть и отвечает на вопрос «когда строка приехала в хранилище», а не «когда её разобрали». В `ods.event` она же служит колонкой версии `ReplacingMergeTree` и схлопывает -повтор доставки. У `ods.order_snapshot` версию задаёт `updated_at` источника; +повтор доставки. У `ods.order` версию задаёт `updated_at` источника; `_load_ts` только показывает, когда конкретная строка приехала. `created_at` и `updated_at` заказа к служебным колонкам хранилища не относятся. @@ -181,9 +181,9 @@ snake_case-имена и расшифровку кодов. ## Версии заказов -`ods.order_snapshot_rep` хранит принятые версии в +`ods.order_rep` хранит принятые версии в `ReplacingMergeTree(updated_at)`: ключ сортировки — `order_id`, партиция — -`toDate(created_at)`. `ods.order_snapshot_dist` шардирует по +`toDate(created_at)`. `ods.order_dist` шардирует по `cityHash64(order_id)`. Все версии заказа лежат в одной партиции, чтобы их могли схлопывать фоновые слияния, и на одном шарде, чтобы распределённый `FINAL` выбрал одного победителя. @@ -414,7 +414,7 @@ D0 и к реальному календарю не привязана; паке kafka_offset)`: смотрят такую таблицу от класса, а внутри класса — по координатам доставки. -`ods.order_snapshot_errors` держит тот же диагностический минимум и `_load_id` +`ods.order_errors` держит тот же диагностический минимум и `_load_id` запуска. У заказов три класса по приоритету: `not_an_object`, `keyset_mismatch`, `field_invalid`. Сырой текст остаётся рядом, поэтому класс не разрастается до имени отдельного поля. Точная граница приёма — в @@ -534,15 +534,15 @@ Airflow читает и собирает эти файлы штатным шаб | `ods` | `ods.event_v` | актуальная версия события с полями источника | | `ods` | `ods.event_errors_rep` / `_dist` | строки, не прошедшие строгий приём | | `ods` | `ods.event_mv`, `ods.event_errors_mv` | разбор сырья в событие и в ошибки | -| `ods` | `ods.order_snapshot_rep` / `_dist` | типизированные версии заказа | +| `ods` | `ods.order_rep` / `_dist` | типизированные версии заказа | | `ods` | `ods.order_v` | текущая версия заказа на языке источника | -| `ods` | `ods.order_snapshot_errors_rep` / `_dist` | строки слепка, не прошедшие строгий приём | +| `ods` | `ods.order_errors_rep` / `_dist` | строки слепка, не прошедшие строгий приём | | `dic` | `dic.products_file` | чтец CSV-каталога, общего с генератором | | `dic` | `dic.products` | словарь товаров поверх подложки | Матвью разбора у заказов нет: срез сырья раскладывают по этим двум целям два -`INSERT SELECT` шага `parse_batch` из файлов `sql/ods/order_snapshot_load.sql` -и `sql/ods/order_snapshot_errors_load.sql`. +`INSERT SELECT` шага `parse_batch` из файлов `sql/ods/order_load.sql` и +`sql/ods/order_errors_load.sql`. Слой DDS и слой DM появляются на следующих этапах; их состав задан разделом 7 мастер-спеки и переносится сюда по мере постройки. @@ -606,14 +606,13 @@ DDL-словарь с файловым источником внутри `user_f `_load_id`, пересечение ветвей — ноль. Все восемь строк брака — версии одного заказа с отрицательным `total`; это дефект генератора, разобранный в [спецификации приёма](orders/ingestion.md), «Что проверено». На управляемой -порции из пяти строк (две версии -одного заказа плюс по одной строке каждого класса брака) годные ушли в -`ods.order_snapshot`, брак — в `ods.order_snapshot_errors` с ожидаемыми -классами. Обе версии легли в одну партицию и на один шард; `FINAL` через -распределённую таблицу вернул одну строку — ту, у которой `updated_at` позже, — -тогда как физический счёт показывал две. Повтор задачи с тем же `_load_id` -строку в `ods.order_v` и её `_load_ts` не изменил, а таблица ошибок записала -тот же брак второй раз, как и обещано спецификацией приёма. +порции из пяти строк (две версии одного заказа плюс по одной строке каждого +класса брака) годные ушли в `ods.order`, брак — в `ods.order_errors` с +ожидаемыми классами. Обе версии легли в одну партицию и на один шард; `FINAL` +через распределённую таблицу вернул одну строку — ту, у которой `updated_at` +позже, — тогда как физический счёт показывал две. Повтор задачи с тем же +`_load_id` строку в `ods.order_v` и её `_load_ts` не изменил, а таблица ошибок +записала тот же брак второй раз, как и обещано спецификацией приёма. Отдельно измерено, что **сводка вставки не годится в счётчики строк**: у запроса с `WHERE` `read_rows` считает прочитанное с диска, а не подошедшее, — diff --git a/docs/research/2026-08-16-order-snapshot-wire-format.md b/docs/research/2026-08-16-order-snapshot-wire-format.md index ce2db32..33fa3bf 100644 --- a/docs/research/2026-08-16-order-snapshot-wire-format.md +++ b/docs/research/2026-08-16-order-snapshot-wire-format.md @@ -64,7 +64,7 @@ Debezium проводит ту же границу внутри одного с `created_at` и `updated_at` — аудит строки источника. Поэтому `toDate(created_at)` в [спецификации приёма заказов](../architecture/orders/ingestion.md) -используется как стабильный технический ключ партиции `ods.order_snapshot`: +используется как стабильный технический ключ партиции `ods.order`: это день создания строки источника, а не доказательство дня бизнес-события. diff --git a/docs/specs/2026-07-30-stand-v2-realism.md b/docs/specs/2026-07-30-stand-v2-realism.md index 77e715d..d99f5d8 100644 --- a/docs/specs/2026-07-30-stand-v2-realism.md +++ b/docs/specs/2026-07-30-stand-v2-realism.md @@ -219,7 +219,7 @@ Ecommerce (заполнены только у торговых событий): Обоснование и проверка разбора — в [исследовании формата](../research/2026-08-16-order-snapshot-wire-format.md). -- `ods.order_snapshot` принимает версии заказа в +- `ods.order` принимает версии заказа в `ReplacingMergeTree(updated_at)`: `ORDER BY order_id`, партиция по дню неизменного `created_at`, шардирование по `cityHash64(order_id)`. `snapshot_date` остаётся датой наблюдения источника, но не задаёт публикацию @@ -410,7 +410,7 @@ README. | STG | `stg.hits_raw_kafka`, `stg.hits_raw` + MV | сырые строки событий, Kafka Engine на обеих нодах | | STG | `stg.orders_raw_kafka`, `stg.orders_raw`, без MV | сырые строки слепка; чтец на ноде 1, забирает пакетный шаг | | ODS | `ods.event` (+`_errors`), `ods.event_v` | типизированные версии событий, брак и точное актуальное состояние | -| ODS | `ods.order_snapshot` (+`_errors`), `ods.order_v` | типизированные версии заказов, брак и точное текущее состояние | +| ODS | `ods.order` (+`_errors`), `ods.order_v` | типизированные версии заказов, брак и точное текущее состояние | | DDS | `dds.session` | сборка сессий из событий (наследник `dds.click`) | | DDS | `dds.event_v` | представление над `ods.event`: snake_case-имена, расшифровка кодов `DeviceCategory`; витрины DM читают его, а не ODS напрямую | | DDS | модель заказов | зерно, связи и материализация проектируются на этапе DDS | @@ -428,7 +428,7 @@ README. матвью к нему не привязана, и сырьё забирает шаг, которым управляет Airflow — одно прямое чтение вставляет порцию в `stg.orders_raw` с `_load_id` запуска. Один следующий `task_id` двумя последовательными запросами пишет годные строки -и ошибки из того же среза. `ods.order_snapshot` хранит версии по `updated_at`, +и ошибки из того же среза. `ods.order` хранит версии по `updated_at`, а не публикует партицию `snapshot_date`. Слой сырья остаётся точкой повтора и разбора одной принятой порции. @@ -436,7 +436,7 @@ README. из опорных точек раздела 12 переформулировано под них. Состав служебных колонок задаёт дока хранилища. Спеке важны два следствия: -`ods.event` и `ods.order_snapshot` получают метку загрузки `_load_ts`, но у +`ods.event` и `ods.order` получают метку загрузки `_load_ts`, но у заказа бизнес-версию задаёт `updated_at`; `_load_id` проходит через STG и обе цели ODS. Таблицы `stg.*_raw` хранят метаданные доставки Kafka вместе с именем читавшей ноды — diff --git a/sql/ddl/20-ods-tables.sql b/sql/ddl/20-ods-tables.sql index 825c4c9..3e56a52 100644 --- a/sql/ddl/20-ods-tables.sql +++ b/sql/ddl/20-ods-tables.sql @@ -175,7 +175,7 @@ ENGINE = Distributed('clickstream_cluster', 'ods', 'event_errors_rep', cityHash6 -- -- items остаётся сырым фрагментом JSON: приём проверяет только, что это -- массив. Что внутри позиций — забота DDS, а не границы провода. -CREATE TABLE IF NOT EXISTS ods.order_snapshot_rep ON CLUSTER clickstream_cluster +CREATE TABLE IF NOT EXISTS ods.order_rep ON CLUSTER clickstream_cluster ( order_id String, user_id UInt64, @@ -199,9 +199,9 @@ ORDER BY order_id; -- и выбор здесь не про перекос, а про корректность: только так все версии -- одного заказа попадают на один шард, и FINAL через распределённую таблицу -- выбирает одного победителя, а не по победителю на шард. -CREATE TABLE IF NOT EXISTS ods.order_snapshot_dist ON CLUSTER clickstream_cluster -AS ods.order_snapshot_rep -ENGINE = Distributed('clickstream_cluster', 'ods', 'order_snapshot_rep', cityHash64(order_id)); +CREATE TABLE IF NOT EXISTS ods.order_dist ON CLUSTER clickstream_cluster +AS ods.order_rep +ENGINE = Distributed('clickstream_cluster', 'ods', 'order_rep', cityHash64(order_id)); -- Локальная таблица брака слепка. -- @@ -215,7 +215,7 @@ ENGINE = Distributed('clickstream_cluster', 'ods', 'order_snapshot_rep', cityHas -- field_invalid. Имя провалившегося поля в класс не входит — сырой текст лежит -- рядом, и единичный случай разбирается по нему, без постоянной детализации -- предиката (docs/architecture/orders/ingestion.md). -CREATE TABLE IF NOT EXISTS ods.order_snapshot_errors_rep ON CLUSTER clickstream_cluster +CREATE TABLE IF NOT EXISTS ods.order_errors_rep ON CLUSTER clickstream_cluster ( raw String, error_class LowCardinality(String), @@ -233,6 +233,6 @@ ORDER BY (error_class, kafka_partition, kafka_offset) TTL toDateTime(_load_ts) + INTERVAL 1 MONTH SETTINGS ttl_only_drop_parts = 1; -CREATE TABLE IF NOT EXISTS ods.order_snapshot_errors_dist ON CLUSTER clickstream_cluster -AS ods.order_snapshot_errors_rep -ENGINE = Distributed('clickstream_cluster', 'ods', 'order_snapshot_errors_rep', cityHash64(raw)); +CREATE TABLE IF NOT EXISTS ods.order_errors_dist ON CLUSTER clickstream_cluster +AS ods.order_errors_rep +ENGINE = Distributed('clickstream_cluster', 'ods', 'order_errors_rep', cityHash64(raw)); diff --git a/sql/ddl/30-ods-views.sql b/sql/ddl/30-ods-views.sql index a5d7f29..51ae10c 100644 --- a/sql/ddl/30-ods-views.sql +++ b/sql/ddl/30-ods-views.sql @@ -22,7 +22,7 @@ FROM ods.event_dist FINAL; CREATE VIEW IF NOT EXISTS ods.order_v ON CLUSTER clickstream_cluster AS SELECT * -FROM ods.order_snapshot_dist FINAL; +FROM ods.order_dist FINAL; -- ODS: разбор сырья в событие и в таблицу ошибок. -- diff --git a/sql/ods/order_snapshot_errors_load.sql b/sql/ods/order_errors_load.sql similarity index 95% rename from sql/ods/order_snapshot_errors_load.sql rename to sql/ods/order_errors_load.sql index 597e9ab..eaac128 100644 --- a/sql/ods/order_snapshot_errors_load.sql +++ b/sql/ods/order_errors_load.sql @@ -4,7 +4,7 @@ -- первый совпавший: скаляр проваливает и проверку на объект, и сверку ключей — -- без объявленного порядка он попал бы то в один класс, то в другой. -INSERT INTO ods.order_snapshot_errors_dist +INSERT INTO ods.order_errors_dist {% include "ods/_order_wire_contract.sql" %} SELECT raw, diff --git a/sql/ods/order_snapshot_load.sql b/sql/ods/order_load.sql similarity index 98% rename from sql/ods/order_snapshot_load.sql rename to sql/ods/order_load.sql index 5c5f8a9..5b3612c 100644 --- a/sql/ods/order_snapshot_load.sql +++ b/sql/ods/order_load.sql @@ -4,7 +4,7 @@ -- Разобранные значения берёт assumeNotNull: обнуляемый разбор стоит за -- предикатом, который NULL уже отсёк, и приведение здесь не может упасть. -INSERT INTO ods.order_snapshot_dist +INSERT INTO ods.order_dist {% include "ods/_order_wire_contract.sql" %} SELECT JSONExtractString(raw, 'order_id') AS order_id,