From 20071c565fecca8be12108d1dac4725437a9d813 Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Wed, 26 Aug 2026 17:58:43 -0400 Subject: [PATCH] =?UTF-8?q?feat(dm):=20=D0=B4=D0=BE=D0=B1=D0=B0=D0=B2?= =?UTF-8?q?=D0=BB=D0=B5=D0=BD=D0=B0=20=D1=81=D0=B2=D0=B5=D1=80=D0=BA=D0=B0?= =?UTF-8?q?=20=D0=BF=D0=BE=D0=BA=D1=83=D0=BF=D0=BE=D0=BA=20=D1=81=20=D0=B7?= =?UTF-8?q?=D0=B0=D0=BA=D0=B0=D0=B7=D0=B0=D0=BC=D0=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Зачем: - показать расхождения трекера и бэкенда на зерне заказа - сохранить отложенные покупки наблюдаемыми до следующего слепка - Что: - добавлены таблицы, представление и пересборка purchase_vs_orders - сумма покупки нормализована в Decimal на границе ODS - обновлены договоры и указатели на механику пересборки - Проверка: - `make config-test`, `make lint` - `cd generator && make docs lint typecheck test` - `make rebuild-storage`, явные `dds_transform` и `dm_transform` - `make smoke`, `make check-clickhouse`, `make check-services` --- dags/dm_transform.py | 100 +++++++++++++++++- docs/architecture/dds/event.md | 5 + docs/architecture/dm.md | 34 +++++- docs/architecture/storage.md | 26 +++-- docs/formats/clickstream-event.md | 12 ++- docs/specs/2026-07-30-stand-v2-realism.md | 13 +-- generator/src/clickstream_generator/schema.py | 16 +-- .../src/clickstream_generator/schema_doc.py | 10 +- sql/ddl/20-ods-tables.sql | 2 +- sql/ddl/30-ods-views.sql | 6 +- sql/ddl/55-dm-tables.sql | 47 +++++++- sql/ddl/60-dm-views.sql | 16 +++ sql/dm/purchase_vs_orders_rebuild.sql | 71 +++++++++++++ sql/dm/purchase_vs_orders_replace.sql | 9 ++ sql/dm/purchase_vs_orders_scope.sql | 62 +++++++++++ 15 files changed, 391 insertions(+), 38 deletions(-) create mode 100644 sql/dm/purchase_vs_orders_rebuild.sql create mode 100644 sql/dm/purchase_vs_orders_replace.sql create mode 100644 sql/dm/purchase_vs_orders_scope.sql diff --git a/dags/dm_transform.py b/dags/dm_transform.py index 4915ef5..d2b38a4 100644 --- a/dags/dm_transform.py +++ b/dags/dm_transform.py @@ -1,8 +1,9 @@ """Трансформации слоя DM: сборка стендовых витрин. -Выручка наследует ритм заказов и заменяет дневные партиции изменяемого окна. -Трафик наследует ретроспективность карты идентичностей и пересчитывает всю -историю новой версией. Объем каждого прогона выводится из состояния слоев. +Выручка наследует ритм заказов. Сверка соединяет окно заказов с прожитым +хвостом событий. Обе заменяют дневные партиции. Трафик наследует +ретроспективность карты идентичностей и пересчитывает всю историю новой +версией. Объем каждого прогона выводится из состояния слоев. """ from __future__ import annotations @@ -14,12 +15,41 @@ from pathlib import Path from airflow.exceptions import AirflowException from airflow.providers.clickhousedb.hooks.clickhouse import ClickHouseHook from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator -from airflow.sdk import dag, task, task_group +from airflow.sdk import Variable, dag, task, task_group START_DATE = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC) SQL_ROOT = Path("/opt/airflow/sql") CLICKHOUSE_CONNECTION = "clickhouse_default" +# Та же ось модельного времени, что у DDS. Общего модуля у дагов нет +# намеренно: связь держит переменная Airflow, а не импорт одного дага другим. +WORLD_POSITION = "world_position" + + +def _world_position() -> int: + """Позиция мира для границы прожитых событий сверки. + + Причина отдельного чтения переменной разобрана у одноименной функции в + ``dds_transform.py``. + """ + raw_position = Variable.get(WORLD_POSITION, default=None) + if raw_position is None: + raise AirflowException( + "мир еще не создан: выполните make up и дождитесь world_initialize" + ) + try: + position = int(raw_position) + except (TypeError, ValueError) as error: + raise AirflowException( + f"world_position={raw_position!r} не является номером дня; " + "выполните make rebuild-storage" + ) from error + if position < 1: + raise AirflowException( + f"мир не готов: world_position={position}; " + "выполните make up или make rebuild-storage" + ) + return position def _days(first_day: datetime.date, last_day: datetime.date) -> list[str]: @@ -46,6 +76,9 @@ def replacements(scope: dict[str, object]) -> list[dict[str, object]]: start_date=START_DATE, is_paused_upon_creation=False, max_active_runs=1, + # Каждая задача LocalExecutor - отдельный процесс. Четыре процесса уже + # дали SIGKILL при лимите scheduler 640 МиБ; три - измеренный потолок DDS. + max_active_tasks=3, template_searchpath=str(SQL_ROOT), tags=["dm"], ) @@ -114,6 +147,65 @@ def dm_transform(): revenue_daily() + @task_group(group_id="purchase_vs_orders") + def purchase_vs_orders(): + """Сверка: окно заказов, прожитый хвост и предварительные дни.""" + + @task + def scope() -> dict[str, object]: + """Спросить у обоих источников, какие дни пересобрать.""" + query = (SQL_ROOT / "dm" / "purchase_vs_orders_scope.sql").read_text( + encoding="utf-8" + ) + hook = ClickHouseHook(clickhouse_conn_id=CLICKHOUSE_CONNECTION) + row = hook.get_first(query, parameters={"position": _world_position()}) + if row is None: + raise AirflowException( + "нет принятого слепка заказов или прожитых покупок; " + "дождитесь источников и сначала выполните dds_transform" + ) + first_day, last_day = row + days = _days(first_day, last_day) + logging.info( + "объем пересборки: %s … %s, дней %d", + first_day, + last_day, + len(days), + ) + return { + "first_day": first_day.isoformat(), + "last_day": last_day.isoformat(), + "days": days, + } + + days = scope() + + rebuild = SQLExecuteQueryOperator( + task_id="rebuild", + conn_id=CLICKHOUSE_CONNECTION, + sql="dm/purchase_vs_orders_rebuild.sql", + parameters={ + "load_id": "{{ run_id }}", + "first_day": days["first_day"], + "last_day": days["last_day"], + }, + split_statements=True, + do_xcom_push=False, + ) + + replace = SQLExecuteQueryOperator.partial( + task_id="replace", + conn_id=CLICKHOUSE_CONNECTION, + sql="dm/purchase_vs_orders_replace.sql", + do_xcom_push=False, + # Причина единицы разобрана у revenue_daily.replace выше. + max_active_tis_per_dag=1, + ).expand_kwargs(replacements(days)) + + rebuild >> replace + + purchase_vs_orders() + SQLExecuteQueryOperator( task_id="daily_traffic", conn_id=CLICKHOUSE_CONNECTION, diff --git a/docs/architecture/dds/event.md b/docs/architecture/dds/event.md index 12836b0..a7cd40d 100644 --- a/docs/architecture/dds/event.md +++ b/docs/architecture/dds/event.md @@ -45,6 +45,11 @@ своё имя здесь создало бы третий синоним. `client_timezone` тоже остаётся — это атрибут посетителя, а не линза времени. +`purchase_revenue` сохраняет смысл поля источника. Тип денег нормализуется в +ODS; точный тип и проверка описаны в разделе +[«Актуальные события»](../storage.md#актуальные-события). DDS только +переименовывает уже нормализованную колонку. + ## Расшифровки Одна: `device_category` — коды 1–4 в строки `desktop` / `mobile` / diff --git a/docs/architecture/dm.md b/docs/architecture/dm.md index 51c9827..3d6734d 100644 --- a/docs/architecture/dm.md +++ b/docs/architecture/dm.md @@ -70,9 +70,17 @@ знать, какие связки новые, а полный пересчёт карты этот след затирает — восстановление следа оставлено лабе повышенной сложности (issue #149). +У `purchase_vs_orders` два источника и одна непрерывная пересборка. Заказная +сторона задает изменяемое окно последним принятым слепком ODS, событийная — +последний прожитый день с `purchase`. Левая граница возвращает первую дыру и +самый ранний день с `awaiting_order`: предварительная строка уже есть в +витрине, поэтому поиск отсутствующего дня ее не заметит. Правая граница +доходит до более позднего из двух хвостов. Так один прогон догоняет пакетный +разгон длиннее K и не оставляет старое ожидание замороженным. + ## Витрины стенда -Оба публичных представления перечисляют деловые колонки и координаты запуска +Публичные представления перечисляют деловые колонки и координаты запуска `_load_id`, `_load_ts` явно. Новая колонка физической таблицы сама в договор не попадет. @@ -98,6 +106,29 @@ шесть категорий стартового мира близки по объему. Более сложный составной ключ распределил бы эту маленькую таблицу без нового урока. +### `dm.purchase_vs_orders_v` + +Зерно — заказ. Покупки из `dds.event_v` сначала сворачиваются по +`purchase_id[1]`: контракт источника обещает один номер и одну объявленную +сумму, а отдельный счет событий оставляет будущий класс `duplicate_event` +наблюдаемым. Заказы приходят из `dds.order_v`. Обе стороны записаны +подзапросами, затем соединены `FULL OUTER GLOBAL JOIN`: голая левая +Distributed-таблица вернула бы каждый непарный заказ по разу с каждого шарда +([исследование](../research/2026-08-22-distributed-join.md)). + +Непарная сторона остается `NULL` благодаря `join_use_nulls = 1`. При +`join_use_nulls = 0` ClickHouse заполнил бы ее нулем или пустой строкой и скрыл +отсутствие источника. Клиентская `declared_revenue` и серверная `items_total` +сравниваются как `Decimal(18, 2)`. Классы проверяются по порядку: +`awaiting_order` для покупки без заказа, затем `cancelled`, `lost_event`, +`duplicate_event`, `amount_delta`, `match`. Поэтому отмененный заказ без +события остается отмененным, а не потерянным. + +Витрина нарезана по `order_day`, цель и донор шардируются по +`cityHash64(order_id)`. Диапазон собирается в доноре одной вставкой и +заменяется по дням. Первичная сборка охватывает всю историю обоих источников; +следующие прогоны идут по двустороннему ритму, описанному выше. + ### `dm.daily_traffic_v` В строке один день. `visitors` считает разные `client_id` из @@ -163,7 +194,6 @@ Airflow. Партиция — день данных; прогон звена DQ - Ключи шардирования остальных материализованных витрин — вместе с их DDL; правило партиции-донора — в [доке хранилища](storage.md). - Тексты учебных заданий и точный список проверок обоих контуров. -- Механика сверки A+C в SQL — рамка в спеке, разделы 4 и 7. ## Отклонённые варианты diff --git a/docs/architecture/storage.md b/docs/architecture/storage.md index 397eee3..7eacb5e 100644 --- a/docs/architecture/storage.md +++ b/docs/architecture/storage.md @@ -221,6 +221,13 @@ ODS и в таблицу ошибок. Нужен ли он выше ODS, реш `dds.event_v` отвечает за другое: даёт событию язык бизнес-модели, в том числе snake_case-имена и расшифровку кодов. +Типы ODS обычно повторяют контракт выгрузки. Денежный `purchaseRevenue` — +исключение: источник объявляет `Array(Float64)`, а материализованное +представление ODS разбирает числа сырого JSON сразу в +`Array(Decimal(18, 2))`. DDS наследует этот тип без нового преобразования. +Прямой разбор подтвержден документацией ClickHouse через Context7 и запросом +к ClickHouse 26.3; сравнение с сырьем подтвердило: копейки всех покупок целы. + ## Версии заказов `ods.order_rep` хранит принятые версии в @@ -483,9 +490,9 @@ kafka_offset)`: смотрят такую таблицу от класса, а административная операция, роль `etl` права на неё не получает. Деньги каталога живут в трёх единицах: в CSV — целые копейки, словарь отдаёт -`Decimal(18, 2)`, `productPrice` события — целые рубли. Разрыв намеренный, на -нём стоит урок про `Float64` ([описание -выгрузки](../formats/clickstream-event.md)). +`Decimal(18, 2)`, `productPrice` события — целые рубли. Поэтому точную сумму +заказа берут из `purchaseRevenue`, а не складывают округленные цены товаров +([описание выгрузки](../formats/clickstream-event.md)). Устройство обоих объектов — почему такой движок, такая форма пути, такой пользователь и такое окно — расписано в @@ -523,7 +530,7 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр | `40-stg-views.sql` | матвью приёма: чтец в сырьё | | `45-dds-tables.sql` | физические модели заказа, сессии и карты идентичностей, доноры партиций | | `50-dds-views.sql` | публичные договоры сущностей DDS | -| `55-dm-tables.sql` | физические витрины выручки и трафика, донор партиций выручки | +| `55-dm-tables.sql` | физические витрины выручки, сверки и трафика, доноры дневных партиций | | `60-dm-views.sql` | публичные договоры витрин DM | Справочники идут сразу за базами: зона `dic` не зависит ни от одного слоя, и @@ -594,6 +601,9 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр | `dm` | `dm.revenue_daily_rep` / `_dist` | дневная выручка по категориям | | `dm` | `dm.revenue_daily_stage_rep` / `_dist` | таблица-двойник, донор дневных партиций выручки | | `dm` | `dm.revenue_daily_v` | публичный договор дневной выручки | +| `dm` | `dm.purchase_vs_orders_rep` / `_dist` | сверка покупок с заказами, партиция — день заказа | +| `dm` | `dm.purchase_vs_orders_stage_rep` / `_dist` | донор дневных партиций сверки | +| `dm` | `dm.purchase_vs_orders_v` | публичный договор сверки с `NULL` для отсутствующей стороны | | `dm` | `dm.daily_traffic_rep` / `_dist` | версии дневных счетчиков трафика | | `dm` | `dm.daily_traffic_v` | точные текущие счетчики трафика с `FINAL` | | `dic` | `dic.products_file` | чтец CSV-каталога, общего с генератором | @@ -605,8 +615,8 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр В слое DDS построены событие, заказ, сессия и карта идентичностей. Событие остаётся представлением над ODS; заказ и сессия гасят повтор заменой партиции; -карта пишет полный набор пар в версионную таблицу. В DM выручка заменяет -партиции изменяемых дней, а трафик пишет полную историю новой версией. +карта пишет полный набор пар в версионную таблицу. В DM выручка и сверка +заменяют партиции изменяемых дней, а трафик пишет полную историю новой версией. ## Что проверено @@ -849,6 +859,10 @@ Kafka с пустым значением (ноль байт) и запись-н промежуточного `toDecimal64`. Проверено на живом слепке; честность разбора сторожит инвариант бэкенда — сумма «цена × количество» совпала с `items_total` у всех 1662 заказов. +- **`JSONExtract` читает `Decimal` и из числа JSON.** Массив + `purchaseRevenue`, объявленный источником как `Array(Float64)`, разбирается + сразу в `Array(Decimal(18, 2))`. Синтаксис подтвержден через Context7 и на + ClickHouse 26.3; все суммы после ODS совпали с сырьем до копейки. **Сказано по памяти, проверки нет.** Группа пуста: оба утверждения, ждавшие матвью разбора, закрыты опытами при исполнении #43 и переехали выше. diff --git a/docs/formats/clickstream-event.md b/docs/formats/clickstream-event.md index 6ec13dd..0f9028e 100644 --- a/docs/formats/clickstream-event.md +++ b/docs/formats/clickstream-event.md @@ -11,10 +11,12 @@ по элементу на товар. Отдельной сущности «визит» в выгрузке нет — визиты собирают на стороне хранилища, а `VisitID` дан как эталон для самопроверки. -Имена и типы колонок — стороны источника. Хранилище принимает их как есть и -нормализует у себя: то же имя в нашем стиле ждёт в столбце «Нормализованное -имя». Это имя источника, приведённое к snake_case, а не имя атрибута в -модели данных: слой DDS складывает свою модель и называет атрибуты по ней. +Имена и типы колонок описывают сторону источника. Обычно ODS принимает их без +изменений. Исключение — `purchaseRevenue`: источник объявляет Float64, а ODS +хранит деньги в Decimal. Точный локальный тип назван в документации +хранилища. Имя в нашем стиле ждет в столбце «Нормализованное имя». Это имя +источника, приведенное к snake_case, а не имя атрибута в модели данных: слой +DDS складывает свою модель и называет атрибуты по ней. Столбец «Тип numpy» показывает, чем колонка представлена внутри генератора; у массивов это тип элемента. Номер — место колонки в выгрузке: порядок задан контрактом. @@ -90,7 +92,7 @@ | № | Колонка | Тип ClickHouse | Тип numpy | Нормализованное имя | Комментарий | |---|---|---|---|---|---| | 37 | `purchaseID` | `Array(String)` | `object` | `purchase_id` | номер заказа; у события purchase — один элемент | -| 38 | `purchaseRevenue` | `Array(Float64)` | `float64` | `purchase_revenue` | выручка заказа глазами клиента; Float64, как у Метрики — на этом держится урок о расхождениях с бэкендом | +| 38 | `purchaseRevenue` | `Array(Float64)` | `float64` | `purchase_revenue` | выручка заказа глазами клиента; источник объявляет Float64, ODS хранит точный Decimal | | 39 | `purchaseCurrency` | `Array(String)` | `object` | `purchase_currency` | валюта заказа | | 40 | `purchaseCoupon` | `Array(String)` | `object` | `purchase_coupon` | купон заказа, если был применён | | 41 | `productID` | `Array(String)` | `object` | `product_id` | id товаров события | diff --git a/docs/specs/2026-07-30-stand-v2-realism.md b/docs/specs/2026-07-30-stand-v2-realism.md index d6fbaef..a8400ef 100644 --- a/docs/specs/2026-07-30-stand-v2-realism.md +++ b/docs/specs/2026-07-30-stand-v2-realism.md @@ -138,8 +138,9 @@ Ecommerce (заполнены только у торговых событий): | `ecommerce` | String — сырой JSON события, как отдаёт Метрика (кандидат будущей лабы: сырое против разобранного) | `add_to_cart` несёт массивы `product*` с одним товаром; `purchase` — состав -заказа и блок `purchase*`. Выручка у клиента — во Float64, как у Метрики: -это не недосмотр, а часть урока о расхождениях (см. раздел 4). +заказа и блок `purchase*`. Выручка приходит JSON-числом Float64, как у +Метрики. ODS один раз нормализует деньги в `Array(Decimal(18, 2))`, не меняя +формат источника; DDS наследует тип хранения без нового вычисления. ### 1.3 Ключи и движки @@ -210,9 +211,9 @@ Ecommerce (заполнены только у торговых событий): ключей совпадает с порядком полей в таблице контракта, у позиции — `sku`, `qty`, `price`: так байты воспроизводимы без сортировки ключей. Разница с кликстримом намеренна: там значения `purchaseRevenue` приезжают JSON-числами - и разбираются как `Array(Float64)`, а бэкенд передаёт деньги строками для - точного `Decimal`. Это показывает расхождение представлений денег в двух - источниках. + Float64, а бэкенд передает деньги строками. На границе ODS оба источника + становятся `Decimal(18, 2)`: клиентское число разбирается сразу, без + промежуточного Float64 в хранилище. Запись собирает явная функция существующего канонического сериализатора: один словарь с вложенным списком и один `orjson.dumps`, без универсального слоя кодеков. @@ -285,7 +286,7 @@ CSV в репозитории (`data/catalog/products.csv`: `sku`, `name`, `cate под именем `declared_*` там, где атрибуция без трекера невозможна (UTM); слово `declared` в имени — сигнал «это заявка клиента, не деньги отчётности». Оно выучивается на конфликте: суммы не сойдутся, менти сам -раскопает почему (Float64 против Decimal, промокод, доставка, отмены). +раскопает почему (состав позиций, промокод, доставка, отмены). ## 5. Анонимность и склейка идентичностей diff --git a/generator/src/clickstream_generator/schema.py b/generator/src/clickstream_generator/schema.py index a1c560a..d31d3f4 100644 --- a/generator/src/clickstream_generator/schema.py +++ b/generator/src/clickstream_generator/schema.py @@ -12,12 +12,12 @@ - `name` — имя источника, как в облачной выгрузке Метрики; сырой слой хранит его без изменений. -- `clickhouse_type` — тип в хранилище ровно в той записи, в какой его вернёт - `system.columns`: параметры входят в имя типа целиком, без сокращений - (`LowCardinality(String)`, `Array(Float64)`). Сверено 2026-08-01 — - по документации ClickHouse через Context7 и запросом к узлу стенда - (26.3.17.56). Запись важна потому, что по ней человек пишет DDL: тип, - сокращённый здесь, приедет в таблицу сокращённым же. +- `clickhouse_type` — тип колонки у источника в той записи, которой пользуется + ClickHouse: параметры входят в имя целиком, без сокращений + (`LowCardinality(String)`, `Array(Float64)`). Обычно ODS берет его без + изменений. Исключение одно: денежный `purchaseRevenue` приезжает как + Float64, а на границе хранения разбирается прямо в Decimal. Поэтому поле + описывает выгрузку, а локальный тип ODS назван в документации хранилища. - `numpy_dtype` — чем колонка представлена внутри генератора; у массивов это тип элемента. Строки живут в `object`-массивах: numpy-строки фиксированной длины стенду ничего не дают. @@ -371,8 +371,8 @@ COLUMNS: tuple[Column, ...] = ( numpy_dtype="float64", normalized_name="purchase_revenue", group=ColumnGroup.ECOMMERCE, - comment="выручка заказа глазами клиента; Float64, как у Метрики —" - " на этом держится урок о расхождениях с бэкендом", + comment="выручка заказа глазами клиента; источник объявляет Float64," + " ODS хранит точный Decimal", ), Column( name="purchaseCurrency", diff --git a/generator/src/clickstream_generator/schema_doc.py b/generator/src/clickstream_generator/schema_doc.py index befe95c..5dd22c2 100644 --- a/generator/src/clickstream_generator/schema_doc.py +++ b/generator/src/clickstream_generator/schema_doc.py @@ -27,10 +27,12 @@ PREAMBLE = """# Описание выгрузки: событие кликстр по элементу на товар. Отдельной сущности «визит» в выгрузке нет — визиты собирают на стороне хранилища, а `VisitID` дан как эталон для самопроверки. -Имена и типы колонок — стороны источника. Хранилище принимает их как есть и -нормализует у себя: то же имя в нашем стиле ждёт в столбце «Нормализованное -имя». Это имя источника, приведённое к snake_case, а не имя атрибута в -модели данных: слой DDS складывает свою модель и называет атрибуты по ней. +Имена и типы колонок описывают сторону источника. Обычно ODS принимает их без +изменений. Исключение — `purchaseRevenue`: источник объявляет Float64, а ODS +хранит деньги в Decimal. Точный локальный тип назван в документации +хранилища. Имя в нашем стиле ждет в столбце «Нормализованное имя». Это имя +источника, приведенное к snake_case, а не имя атрибута в модели данных: слой +DDS складывает свою модель и называет атрибуты по ней. Столбец «Тип numpy» показывает, чем колонка представлена внутри генератора; у массивов это тип элемента. Номер — место колонки в выгрузке: порядок задан контрактом. diff --git a/sql/ddl/20-ods-tables.sql b/sql/ddl/20-ods-tables.sql index 92a8011..e914116 100644 --- a/sql/ddl/20-ods-tables.sql +++ b/sql/ddl/20-ods-tables.sql @@ -83,7 +83,7 @@ CREATE TABLE IF NOT EXISTS ods.event_rep ON CLUSTER clickstream_cluster GoalsReached Array(UInt32), ParsedParamsKey1 Array(String), purchaseID Array(String), - purchaseRevenue Array(Float64), + purchaseRevenue Array(Decimal(18, 2)), purchaseCurrency Array(String), purchaseCoupon Array(String), productID Array(String), diff --git a/sql/ddl/30-ods-views.sql b/sql/ddl/30-ods-views.sql index 142f18d..6b42c12 100644 --- a/sql/ddl/30-ods-views.sql +++ b/sql/ddl/30-ods-views.sql @@ -154,7 +154,11 @@ SELECT JSONExtract(raw, 'GoalsReached', 'Array(UInt32)') AS GoalsReached, JSONExtract(raw, 'ParsedParamsKey1', 'Array(String)') AS ParsedParamsKey1, JSONExtract(raw, 'purchaseID', 'Array(String)') AS purchaseID, - JSONExtract(raw, 'purchaseRevenue', 'Array(Float64)') AS purchaseRevenue, + -- Выгрузка несет JSON-числа типа Float64, но деньги хранятся в Decimal. + -- JSONExtract разбирает их сразу, без промежуточного двоичного числа: + -- синтаксис подтвержден документацией ClickHouse через 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, JSONExtract(raw, 'productID', 'Array(String)') AS productID, diff --git a/sql/ddl/55-dm-tables.sql b/sql/ddl/55-dm-tables.sql index 6ccd51a..de4ea26 100644 --- a/sql/ddl/55-dm-tables.sql +++ b/sql/ddl/55-dm-tables.sql @@ -1,4 +1,4 @@ --- DM: физические дневные витрины и донор партиций выручки. +-- DM: физические дневные витрины и доноры их партиций. -- Выручка хранится в зерне «день заказа × категория товара». День служит -- единицей пересборки, поэтому совпадает с ключом партиции. @@ -48,6 +48,51 @@ ENGINE = Distributed( cityHash64(product_category) ); +-- Сверка хранит одну строку на заказ. День — единица пересборки: пока рядом +-- живут два источника, строка может перейти из awaiting_order в обычный класс. +CREATE TABLE IF NOT EXISTS dm.purchase_vs_orders_rep ON CLUSTER clickstream_cluster +( + order_day Date, + order_id String, + declared_revenue Nullable(Decimal(18, 2)), + items_total Nullable(Decimal(18, 2)), + status LowCardinality(Nullable(String)), + mismatch_class LowCardinality(String), + _load_id String, + _load_ts DateTime64(3, 'UTC') +) +ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/{table}', '{replica}') +PARTITION BY order_day +ORDER BY (order_day, order_id); + +-- Ключ совпадает с DDS, поэтому строка заказа на всех слоях попадает на один +-- шард. +CREATE TABLE IF NOT EXISTS dm.purchase_vs_orders_dist ON CLUSTER clickstream_cluster +AS dm.purchase_vs_orders_rep +ENGINE = Distributed( + 'clickstream_cluster', + 'dm', + 'purchase_vs_orders_rep', + cityHash64(order_id) +); + +-- Донор повторяет структуру и раскладку цели по правилу, разобранному у +-- revenue_daily_stage_rep и в docs/architecture/storage.md. +CREATE TABLE IF NOT EXISTS dm.purchase_vs_orders_stage_rep ON CLUSTER clickstream_cluster +AS dm.purchase_vs_orders_rep +ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/{table}', '{replica}') +PARTITION BY order_day +ORDER BY (order_day, order_id); + +CREATE TABLE IF NOT EXISTS dm.purchase_vs_orders_stage_dist ON CLUSTER clickstream_cluster +AS dm.purchase_vs_orders_stage_rep +ENGINE = Distributed( + 'clickstream_cluster', + 'dm', + 'purchase_vs_orders_stage_rep', + cityHash64(order_id) +); + -- Трафик пересчитывается целиком: новая связь куки с пользователем меняет -- прошлые дни. Версия запуска позволяет точному представлению скрыть строки -- прежних пересчетов без TRUNCATE и без партиций. diff --git a/sql/ddl/60-dm-views.sql b/sql/ddl/60-dm-views.sql index 4b132c2..f47a130 100644 --- a/sql/ddl/60-dm-views.sql +++ b/sql/ddl/60-dm-views.sql @@ -15,6 +15,22 @@ SELECT _load_ts FROM dm.revenue_daily_dist; +-- Непарные стороны остаются NULL. При join_use_nulls = 0 ClickHouse заполнил +-- бы их нулями и пустыми строками, скрыв отсутствие источника. +-- Служебные колонки называют запуск, который последним заменил день. +CREATE VIEW IF NOT EXISTS dm.purchase_vs_orders_v ON CLUSTER clickstream_cluster +AS +SELECT + order_day, + order_id, + declared_revenue, + items_total, + status, + mismatch_class, + _load_id, + _load_ts +FROM dm.purchase_vs_orders_dist; + -- Полный пересчет пишет новую версию каждого дня. FINAL оставляет текущие -- счетчики и прячет повторы запусков от потребителя. CREATE VIEW IF NOT EXISTS dm.daily_traffic_v ON CLUSTER clickstream_cluster diff --git a/sql/dm/purchase_vs_orders_rebuild.sql b/sql/dm/purchase_vs_orders_rebuild.sql new file mode 100644 index 0000000..f4559c6 --- /dev/null +++ b/sql/dm/purchase_vs_orders_rebuild.sql @@ -0,0 +1,71 @@ +-- DM: сверка покупок трекера с текущими заказами бэкенда. +-- +-- Обе стороны FULL OUTER JOIN — подзапросы. С голой левой Distributed +-- ClickHouse вернул бы непарный заказ по разу с каждого шарда; форма измерена +-- в docs/research/2026-08-22-distributed-join.md. +-- Выбор DROP PARTITION ALL вместо TRUNCATE разобран в соседнем +-- sql/dds/order_rebuild.sql. + +ALTER TABLE dm.purchase_vs_orders_stage_rep ON CLUSTER clickstream_cluster +DROP PARTITION ALL; + +INSERT INTO dm.purchase_vs_orders_stage_dist +( + order_day, + order_id, + declared_revenue, + items_total, + status, + mismatch_class, + _load_id, + _load_ts +) +SELECT + -- У парного заказа день берется у бэкенда; трекер дает его только + -- непарной покупке. + coalesce(order_date, purchase_day), + coalesce(purchase_order_id, backend_order_id), + declared_revenue, + items_total, + status, + multiIf( + backend_order_id IS NULL, 'awaiting_order', + status = 'cancelled', 'cancelled', + purchase_order_id IS NULL, 'lost_event', + purchase_count > 1, 'duplicate_event', + declared_revenue != items_total, 'amount_delta', + 'match' + ), + {load_id:String}, + now64(3, 'UTC') +FROM +( + -- Контракт purchase несет один номер и одну объявленную сумму в массивах. + -- Бизнес-дубль повторяет оба значения, поэтому count хранится отдельно, + -- а сумма берется одна — складывать дубли значило бы выдумать дельту. + SELECT + purchase_id[1] AS purchase_order_id, + any(event_date) AS purchase_day, + any(purchase_revenue[1]) AS declared_revenue, + count() AS purchase_count + FROM dds.event_v + WHERE event_type = 'purchase' + AND event_date BETWEEN {first_day:Date} AND {last_day:Date} + GROUP BY purchase_order_id +) AS purchases +GLOBAL FULL OUTER JOIN +( + SELECT + order_id AS backend_order_id, + order_date, + items_total, + status + FROM dds.order_v + WHERE order_date BETWEEN {first_day:Date} AND {last_day:Date} +) AS orders ON purchase_order_id = backend_order_id +-- NULL обозначает отсутствующую сторону. При join_use_nulls = 0 ClickHouse +-- заполнил бы ее нулем или пустой строкой и скрыл отсутствие источника. +-- Вставка идет через ключ цели: источники разложены по другим ключам. +SETTINGS join_use_nulls = 1, + distributed_foreground_insert = 1, + parallel_distributed_insert_select = 0; diff --git a/sql/dm/purchase_vs_orders_replace.sql b/sql/dm/purchase_vs_orders_replace.sql new file mode 100644 index 0000000..44ec3d9 --- /dev/null +++ b/sql/dm/purchase_vs_orders_replace.sql @@ -0,0 +1,9 @@ +-- DM: подмена одного дня сверки собранной партицией донора. +-- +-- День для PARTITION берет purchase_vs_orders_scope.sql. Выбор Jinja, +-- атомарность и повтор разобраны в sql/dds/order_replace.sql. +-- +-- Пустая партиция законна: если источник больше не показывает строк дня, +-- замена убирает этот день из витрины. +ALTER TABLE dm.purchase_vs_orders_rep ON CLUSTER clickstream_cluster +REPLACE PARTITION '{{ params.day }}' FROM dm.purchase_vs_orders_stage_rep; diff --git a/sql/dm/purchase_vs_orders_scope.sql b/sql/dm/purchase_vs_orders_scope.sql new file mode 100644 index 0000000..e52cc16 --- /dev/null +++ b/sql/dm/purchase_vs_orders_scope.sql @@ -0,0 +1,62 @@ +-- DM: объем пересборки сверки покупок с заказами. +-- +-- У источников разные правые границы. Заказы называют изменяемое окно ярлыком +-- последнего принятого слепка. События заканчиваются последним прожитым днем, +-- который выводится из позиции мира. Одна граница другую не подменяет. +-- +-- Левая граница возвращает не только пропуски, но и самый ранний +-- awaiting_order. Такой день уже есть в витрине, поэтому обычный поиск дыры +-- его не увидит. Иначе пакетный разгон длиннее K навсегда оставил бы прежний +-- хвост предварительным. +-- +-- Файл заканчивается без точки с запятой: почему — sql/dds/order_scope.sql. +WITH + (SELECT min(event_date) FROM dds.event_v) AS first_event_day, + first_event_day + ({position:UInt16} - 1) AS last_lived_day, + (SELECT max(snapshot_date) FROM ods.order_dist) AS last_snapshot, + ( + SELECT min(toDate(toTimeZone(created_at, 'Europe/Samara'))) + FROM ods.order_dist + WHERE snapshot_date = last_snapshot + ) AS window_start, + ( + SELECT max(toDate(toTimeZone(created_at, 'Europe/Samara'))) + FROM ods.order_dist + WHERE snapshot_date = last_snapshot + ) AS window_end, + ( + SELECT maxOrNull(event_date) + FROM dds.event_v + WHERE event_type = 'purchase' AND event_date <= last_lived_day + ) AS last_purchase_day, + ( + SELECT minOrNull(day) + FROM ( + SELECT order_date AS day + FROM dds.order_v + UNION DISTINCT + SELECT event_date AS day + FROM dds.event_v + WHERE event_type = 'purchase' AND event_date <= last_lived_day + ) + WHERE day GLOBAL NOT IN ( + SELECT order_day FROM dm.purchase_vs_orders_dist + ) + ) AS first_missing_day, + ( + SELECT minOrNull(order_day) + FROM dm.purchase_vs_orders_dist + WHERE mismatch_class = 'awaiting_order' + ) AS first_provisional_day +SELECT + least( + window_start, + coalesce(first_missing_day, window_start), + coalesce(first_provisional_day, window_start) + ) AS first_day, + greatest(window_end, last_purchase_day) AS last_day +-- Без принятого слепка неизвестно окно заказов. Без готового DDS или прожитой +-- покупки донор был бы пуст, поэтому подменять партиции еще нельзя. +WHERE (SELECT count() FROM ods.order_dist) > 0 + AND (SELECT count() FROM dds.order_v) > 0 + AND last_purchase_day IS NOT NULL