diff --git a/dags/dds_transform.py b/dags/dds_transform.py index 8035565..8cccc73 100644 --- a/dags/dds_transform.py +++ b/dags/dds_transform.py @@ -1,18 +1,18 @@ -"""Трансформации слоя DDS: сборка модели заменой дневных партиций. +"""Трансформации слоя DDS: сборка моделей. -Даг собирает модель поверх принятого ODS. Сущностей у него две — заказ и -сессия, зависимости по данным между ними нет; карта идентичностей встанет -соседней группой задач. +Заказ и сессия собираются заменой дневных партиций. Карта идентичностей +пересчитывается целиком после заказа, потому что читает его текущую модель. Расписания у дага нет и параметров тоже: объём каждого прогона он выводит сам из состояния слоёв и позиции мира, поэтому один и тот же запуск одинаково верен из триггера дневного конвейера, с кнопки менти и после падения (docs/architecture/etl.md, «Объём прогона»). -Повтор прогона гасит не колонка версии, а замена партиции целиком: день -собирается в таблицу-двойник и подменяет собой партицию цели. Почему слой -устроен так — sql/ddl/45-dds-tables.sql; механика самой подмены — -sql/dds/order_replace.sql. +Заказ и сессия гасят повтор заменой партиции целиком: день собирается в +таблицу-двойник и подменяет собой партицию цели. Карта пишет полный набор пар +новой версией, а точное чтение даёт FINAL. Замена партиции разобрана в +sql/dds/order_replace.sql; точное чтение карты описано в +sql/ddl/50-dds-views.sql. """ from __future__ import annotations @@ -229,8 +229,18 @@ def dds_transform(): rebuild >> replace - order() + orders = order() session() + identity_map = SQLExecuteQueryOperator( + task_id="identity_map", + conn_id=CLICKHOUSE_CONNECTION, + sql="dds/identity_map_rebuild.sql", + parameters={"load_id": "{{ run_id }}"}, + do_xcom_push=False, + ) + + orders >> identity_map + dds_transform() diff --git a/docs/architecture/storage.md b/docs/architecture/storage.md index 1933f86..54558fe 100644 --- a/docs/architecture/storage.md +++ b/docs/architecture/storage.md @@ -587,6 +587,8 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр | `dds` | `dds.session_rep` / `_dist` | визит, нарезанный слоем; партиция — день сессии | | `dds` | `dds.session_stage_rep` / `_dist` | двойник сессии: в нём день собирается перед заменой партиции | | `dds` | `dds.session_v` | публичный договор сессии | +| `dds` | `dds.identity_map_rep` / `_dist` | известная пара куки с пользователем; таблицы полного пересчёта | +| `dds` | `dds.identity_map_v` | точный публичный договор карты идентичностей с `FINAL` | | `dic` | `dic.products_file` | чтец CSV-каталога, общего с генератором | | `dic` | `dic.products` | словарь товаров поверх подложки | @@ -594,11 +596,11 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр `INSERT SELECT` одной задачи `parse_batch` из файла `sql/ods/order_parse.sql`. -В слое DDS построены событие, заказ и сессия, и устроены они намеренно -по-разному: событие остаётся представлением над ODS, у заказа и сессии — -физические модели. Карту идентичностей и слой DM принесут следующие тикеты -этапа. Их состав задан разделом 7 мастер-спеки и переносится сюда по мере -постройки. +В слое DDS построены событие, заказ, сессия и карта идентичностей. Событие +остаётся представлением над ODS; заказ и сессия гасят повтор заменой партиции; +карта пишет полный набор пар в версионную таблицу. Слой DM принесут следующие +тикеты этапа. Его состав задан разделом 7 мастер-спеки и переносится сюда по +мере постройки. ## Что проверено @@ -818,6 +820,11 @@ Kafka с пустым значением (ноль байт) и запись-н двойник разложила 1662 заказа строго по ключу: шард 1 получил только чётные `cityHash64(order_id)`, шард 2 — только нечётные. +**Проверка карты идентичностей 26 августа 2026 года (#153).** Context7 +подтвердил версию в `ReplacingMergeTree` и точное чтение через `FINAL`. Два +полных прогона оставили 3180 физических строк и 1590 пар в `_v`; все пары +подтверждены покупкой и заказом, а строк на чужом шарде нет. + Тем же днём измерены ещё три поведения ClickHouse. - **Параметр в `ALTER … PARTITION` не подставляется.** Запрос с `{day:Date}` в diff --git a/sql/ddl/45-dds-tables.sql b/sql/ddl/45-dds-tables.sql index 8ea7e8e..662591b 100644 --- a/sql/ddl/45-dds-tables.sql +++ b/sql/ddl/45-dds-tables.sql @@ -1,11 +1,7 @@ --- DDS: физические модели слоя и доноры их партиций. +-- DDS: физические модели слоя и доноры партиций. -- --- Идемпотентность здесь устроена иначе, чем в приёме, и это главный урок обеих --- сущностей файла. В ODS повтор гасит колонка версии: ReplacingMergeTree по --- snapshot_date оставляет строку позднейшего слепка. Здесь колонки --- версии нет вовсе — повтор гасит замена партиции целиком: день собирается --- заново и подменяется одним атомарным движением. Поэтому строка таблицы --- ровно одна на своё зерно, а чтению не нужен FINAL. +-- Заказ и сессия гасят повтор заменой дневной партиции, без колонки версии. +-- Карта идентичностей пересчитывается целиком и пишет новую версию пары. -- Локальная таблица заказа. -- @@ -170,3 +166,33 @@ ORDER BY (client_id, started_at); CREATE TABLE IF NOT EXISTS dds.session_stage_dist ON CLUSTER clickstream_cluster AS dds.session_stage_rep ENGINE = Distributed('clickstream_cluster', 'dds', 'session_stage_rep', cityHash64(client_id)); + +-- DDS: известная связь куки трекера с пользователем магазина. +-- +-- Полный запуск снова пишет все известные пары. Физические повторы до +-- фонового слияния допустимы: _load_ts выбирает последнюю версию пары. +-- Партиции нет, потому что карта пересчитывается целиком и её зерно не +-- связано с днём. +CREATE TABLE IF NOT EXISTS dds.identity_map_rep ON CLUSTER clickstream_cluster +( + client_id UInt64, + user_id UInt64, + _load_id String, + _load_ts DateTime64(3, 'UTC') +) +ENGINE = ReplicatedReplacingMergeTree( + '/clickhouse/tables/{shard}/{database}/{table}', + '{replica}', + _load_ts +) +ORDER BY (client_id, user_id); + +-- Карта лежит по ключу будущего соединения с событиями и сессиями. +CREATE TABLE IF NOT EXISTS dds.identity_map_dist ON CLUSTER clickstream_cluster +AS dds.identity_map_rep +ENGINE = Distributed( + 'clickstream_cluster', + 'dds', + 'identity_map_rep', + cityHash64(client_id) +); diff --git a/sql/ddl/50-dds-views.sql b/sql/ddl/50-dds-views.sql index d8eba06..86184c0 100644 --- a/sql/ddl/50-dds-views.sql +++ b/sql/ddl/50-dds-views.sql @@ -134,3 +134,17 @@ SELECT _load_id, _load_ts FROM dds.session_dist; + +-- DDS: точная известная связь куки с пользователем магазина. +-- +-- Полный пересчёт оставляет физические версии одной пары до слияния частей. +-- FINAL прячет это устройство от читателя; явный список колонок не даёт +-- служебным изменениям таблицы случайно расширить договор. +CREATE VIEW IF NOT EXISTS dds.identity_map_v ON CLUSTER clickstream_cluster +AS +SELECT + client_id, + user_id, + _load_id, + _load_ts +FROM dds.identity_map_dist FINAL; diff --git a/sql/dds/identity_map_rebuild.sql b/sql/dds/identity_map_rebuild.sql new file mode 100644 index 0000000..addb6c4 --- /dev/null +++ b/sql/dds/identity_map_rebuild.sql @@ -0,0 +1,46 @@ +-- DDS: полный пересчёт известных пар куки с пользователем. +-- +-- События разложены по client_id; заказы по order_id. Ключи соединения не +-- ко-локированы, поэтому правило слоя требует явный GLOBAL. С подзапросами с +-- обеих сторон ClickHouse 26.3 выполняет соединение один раз на инициаторе +-- (docs/research/2026-08-22-distributed-join.md). + +INSERT INTO dds.identity_map_dist +( + client_id, + user_id, + _load_id, + _load_ts +) +SELECT + client_id, + user_id, + {load_id:String}, + now64(3, 'UTC') +FROM +( + SELECT DISTINCT + purchase.client_id, + orders.user_id + FROM + ( + SELECT + client_id, + -- По договору события purchase_id содержит ровно один номер + -- заказа: одно подтверждение оформляет один заказ (спека, раздел 4). + purchase_id[1] AS order_id + FROM dds.event_v + WHERE event_type = 'purchase' + ) AS purchase + GLOBAL INNER JOIN + ( + SELECT + order_id, + user_id + FROM dds.order_v + ) AS orders ON purchase.order_id = orders.order_id +) +-- В этой форме ClickHouse 26.3 уже пишет через инициатор. Явный ноль фиксирует +-- раскладку по ключу цели, если форма запроса или план изменятся (капкан #137). +SETTINGS distributed_foreground_insert = 1, + parallel_distributed_insert_select = 0;