From da0477f0f1f8b8e29777745a6201b8baf04fa6bd Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Wed, 26 Aug 2026 13:55:26 -0400 Subject: [PATCH 1/2] =?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=D1=8B=20=D0=B2=D0=B8=D1=82=D1=80=D0=B8=D0=BD?= =?UTF-8?q?=D1=8B=20=D0=B2=D1=8B=D1=80=D1=83=D1=87=D0=BA=D0=B8=20=D0=B8=20?= =?UTF-8?q?=D1=82=D1=80=D0=B0=D1=84=D0=B8=D0=BA=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Зачем: - Показать два ритма пересборки агрегатов на готовых витринах DM. Что: - Добавлены физические модели и публичные договоры revenue_daily и daily_traffic. - Добавлен dm_transform с оконной заменой выручки и полным пересчетом трафика. - Отрицательный опыт локального соединения через публичные _v зафиксирован; выбран явный GLOBAL LEFT JOIN. - Замены партиций сериализованы после подтвержденного OOM при трех одновременных задачах. - Обновлены архитектурные документы и мастер-спека. Проверка: - make lint, make config-test, make smoke, make check-clickhouse, make check-services. - Чистая сборка world_recreate, dds_transform и dm_transform; меры сверены с прямыми агрегатами. --- dags/dm_transform.py | 128 ++++++++++++++++++++++ docs/architecture/dds/event.md | 6 +- docs/architecture/dm.md | 62 +++++++++-- docs/architecture/storage.md | 16 ++- docs/specs/2026-07-30-stand-v2-realism.md | 15 ++- sql/ddl/00-databases.sql | 3 + sql/ddl/55-dm-tables.sql | 78 +++++++++++++ sql/ddl/60-dm-views.sql | 28 +++++ sql/dm/daily_traffic_rebuild.sql | 27 +++++ sql/dm/revenue_daily_rebuild.sql | 55 ++++++++++ sql/dm/revenue_daily_replace.sql | 8 ++ sql/dm/revenue_daily_scope.sql | 35 ++++++ 12 files changed, 436 insertions(+), 25 deletions(-) create mode 100644 dags/dm_transform.py create mode 100644 sql/ddl/55-dm-tables.sql create mode 100644 sql/ddl/60-dm-views.sql create mode 100644 sql/dm/daily_traffic_rebuild.sql create mode 100644 sql/dm/revenue_daily_rebuild.sql create mode 100644 sql/dm/revenue_daily_replace.sql create mode 100644 sql/dm/revenue_daily_scope.sql diff --git a/dags/dm_transform.py b/dags/dm_transform.py new file mode 100644 index 0000000..066b73f --- /dev/null +++ b/dags/dm_transform.py @@ -0,0 +1,128 @@ +"""Трансформации слоя DM: сборка стендовых витрин. + +Выручка наследует ритм заказов и заменяет дневные партиции изменяемого окна. +Трафик наследует ретроспективность карты идентичностей и пересчитывает всю +историю новой версией. Объем каждого прогона выводится из состояния слоев. + +Форма DAG и динамическая замена сверены с публичным API Airflow 3 через +Context7; DDL и REPLACE PARTITION — с документацией ClickHouse. +""" + +from __future__ import annotations + +import datetime +import logging +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 + +START_DATE = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC) + +SQL_ROOT = Path("/opt/airflow/sql") +CLICKHOUSE_CONNECTION = "clickhouse_default" + + +def _days(first_day: datetime.date, last_day: datetime.date) -> list[str]: + """Дни замены строками от левой до правой границы включительно.""" + return [ + (first_day + datetime.timedelta(days=shift)).isoformat() + for shift in range((last_day - first_day).days + 1) + ] + + +@task +def replacements(scope: dict[str, object]) -> list[dict[str, object]]: + """Разложить дни в аргументы размноженной задачи замены. + + Отдельный словарь нужен каждому экземпляру ``expand_kwargs``; тот же + прием разобран в ``dds_transform.py`` на замене партиций заказов. + """ + return [{"params": {"day": day}} for day in scope["days"]] + + +@dag( + dag_id="dm_transform", + schedule=None, + start_date=START_DATE, + is_paused_upon_creation=False, + max_active_runs=1, + max_active_tasks=3, + template_searchpath=str(SQL_ROOT), + tags=["dm"], +) +def dm_transform(): + """Пересобрать стендовые витрины по текущему состоянию DDS.""" + + @task_group(group_id="revenue_daily") + def revenue_daily(): + """Выручка: изменяемое окно заказов и пропущенная история.""" + + @task + def scope() -> dict[str, object]: + """Спросить у хранилища, какие дни пересобирает запуск.""" + query = (SQL_ROOT / "dm" / "revenue_daily_scope.sql").read_text( + encoding="utf-8" + ) + hook = ClickHouseHook(clickhouse_conn_id=CLICKHOUSE_CONNECTION) + row = hook.get_first(query) + if row is None: + raise AirflowException( + "нет слепка заказов в ODS или готовых заказов в DDS" + ) + 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/revenue_daily_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/revenue_daily_replace.sql", + do_xcom_push=False, + # Каждая замена поднимает отдельный процесс работника. Три + # одновременные задачи выбили его по памяти на первом прогоне; + # самим дням параллельность ничего не дает. + max_active_tis_per_dag=1, + ).expand_kwargs(replacements(days)) + + rebuild >> replace + + revenue_daily() + + SQLExecuteQueryOperator( + task_id="daily_traffic", + conn_id=CLICKHOUSE_CONNECTION, + sql="dm/daily_traffic_rebuild.sql", + parameters={"load_id": "{{ run_id }}"}, + do_xcom_push=False, + ) + + +dm_transform() diff --git a/docs/architecture/dds/event.md b/docs/architecture/dds/event.md index a77f8a5..12836b0 100644 --- a/docs/architecture/dds/event.md +++ b/docs/architecture/dds/event.md @@ -89,10 +89,8 @@ - Сборка сессий: ключи нарезки (`client_id`, `event_time`, `event_date`), тип события, страница, UTM, паспорт куки — категория устройства приходит уже расшифрованной. -- Витрины `dm.events_enriched_v`, `dm.top_pages_daily_v`, - `dm.utm_effectiveness_v` (`declared_revenue` — из `purchase_revenue`), - `dm.daily_traffic_v` (пара «посетители / известные» через - `dds.identity_map_v`). +- Витрины `dm.events_enriched_v`, `dm.top_pages_daily_v` и + `dm.utm_effectiveness_v` (`declared_revenue` берется из `purchase_revenue`). - Событийная сторона сверки `dm.purchase_vs_orders_v`: `purchase_id`, `purchase_revenue`, `event_date`. - DQ-сверка сессий с эталоном читает не его, а `ods.event_v` — diff --git a/docs/architecture/dm.md b/docs/architecture/dm.md index fff03cf..51c9827 100644 --- a/docs/architecture/dm.md +++ b/docs/architecture/dm.md @@ -31,8 +31,8 @@ витрины, на него ссылаются учебные задания; - `purchase_vs_orders` — сверка A+C с классами расхождений; зерно — заказ, шардирование сохраняет ко-локацию заказов по `cityHash64(order_id)`; -- `daily_traffic` — пара «посетители» / «известные пользователи», локальное - обогащение [картой идентичностей](dds/identity.md); +- `daily_traffic` — пара «посетители» / «известные пользователи», обогащение + [картой идентичностей](dds/identity.md); - `dq_summary` — сводка проверок качества (ниже). Остальные витрины спеки — учебные задания: менти строит их сам по ТЗ с @@ -55,8 +55,8 @@ - **от событий и сессий** — день замерзает сразу; пересборка догоняет хвост несобранных дней заменой дневных партиций; -- **от заказов** — день дышит окном K и закрывается; пересборка заменяет - партиции окна, дни за ним закрыты по построению; +- **от заказов** — день дышит, пока входит в последний слепок; пересборка + заменяет партиции этого отрезка, дни за ним закрыты по построению; - **от карты идентичностей** — не замерзает никогда: склейка ретроспективна, покупка сегодня делает куку известной во всех её прошлых днях, а карта пересчитывается целиком. @@ -64,12 +64,58 @@ Поэтому `daily_traffic` наследует у своего источника и ритм, и механику: полный пересчёт всех дней каждый прогон в `ReplacingMergeTree(_load_ts)`, как у самой карты. Дневных партиций у неё нет: партиция — единица замены, -а эта витрина заменяется только целиком. Пересчёт дешёв: -витрина читает сессии и карту локальным соединением по ключу ко-локации. +а эта витрина заменяется только целиком. На стенде такая работа остается +быстрой; витрина читает сессии и карту через их публичные договоры. Пересчитывать точечно только затронутые дни не из чего: для этого надо знать, какие связки новые, а полный пересчёт карты этот след затирает — восстановление следа оставлено лабе повышенной сложности (issue #149). +## Витрины стенда + +Оба публичных представления перечисляют деловые колонки и координаты запуска +`_load_id`, `_load_ts` явно. Новая колонка физической таблицы сама в договор не +попадет. + +### `dm.revenue_daily_v` + +Зерно: день заказа × категория товара. Витрина читает только оплаченные +заказы из `dds.order_v`, синхронно разворачивает три массива позиций и узнает +категорию артикула через `dic.products`. Один заказ с несколькими позициями +категории считается в `orders` один раз. `units` складывает количество единиц, +`revenue` складывает цену позиции × количество. Скидка, доставка и итог заказа +между категориями не делятся. `aov` равен `revenue / orders` в том же срезе и +хранит два знака после запятой. + +Первый запуск строит всю доступную историю. Следующие берут границы дней из +последнего слепка ODS: только источник знает, какие даты еще может изменить. +Пропущенный оплаченный день расширяет левую границу, чтобы частичный отказ +не оставил дыру после следующего прогона. Агрегат при этом читает DDS; ODS +отвечает только на вопрос об объеме. + +`dm.revenue_daily_rep` нарезана по `report_date`; день собирается в +таблице-доноре и заменяется через `REPLACE PARTITION`. Цель и донор шардируются +по `cityHash64(product_category)`. Кроме дня, у витрины только категория, а +шесть категорий стартового мира близки по объему. Более сложный составной ключ +распределил бы эту маленькую таблицу без нового урока. + +### `dm.daily_traffic_v` + +В строке один день. `visitors` считает разные `client_id` из +`dds.session_v`; `known_users` считает разные `user_id` после левого соединения +с полной текущей `dds.identity_map_v`. Анонимная сессия остается в витрине. +Несколько кук одного человека увеличивают `visitors`, но дают одного +`known_user`, поэтому `known_users <= visitors`. + +Сессии и карта лежат по `client_id`, но публичные `_v` скрывают +`Distributed`-таблицы от локальной подстановки. На ClickHouse 26.3 +`EXPLAIN PLAN` показал чтение обоих шардов на инициаторе: настройка +`distributed_product_mode = 'local'` локального соединения не дала. Поэтому +витрина использует явный `GLOBAL LEFT JOIN` через публичные договоры, а не +подменяет их физическими таблицами. Это дороже по сети, зато граница слоев +остается явной. Новая связь куки меняет прошлые дни, поэтому каждый запуск +пишет полную историю новой версией в `ReplacingMergeTree(_load_ts)`, а +публичное представление выбирает текущие строки через `FINAL`. + ## `dm.dq_summary` Зерно — «день данных × проверка»: строка отвечает, каково качество дня @@ -113,7 +159,7 @@ Airflow. Партиция — день данных; прогон звена DQ ## Остаётся нарезке -- Имена задач `dm_transform`, звена DQ и дага самопроверки. +- Имена звена DQ и дага самопроверки. - Ключи шардирования остальных материализованных витрин — вместе с их DDL; правило партиции-донора — в [доке хранилища](storage.md). - Тексты учебных заданий и точный список проверок обоих контуров. @@ -136,7 +182,7 @@ Airflow. Партиция — день данных; прогон звена DQ расшифровывать. Красное «ещё не построено» честнее. - **Самопроверка пишет в `dq_summary`.** Попытки менти смешались бы с историей качества конвейера. -- **Окно K для `daily_traffic` — прошлое замерзает.** Дни за окном хранили +- **Окно заказов для `daily_traffic` — прошлое замерзает.** Дни за окном хранили бы число известных на момент закрытия — витрина тихо расходится с картой и протаскивает обратно «знание на дату», отвергнутое развилкой карты (#133). diff --git a/docs/architecture/storage.md b/docs/architecture/storage.md index 54558fe..397eee3 100644 --- a/docs/architecture/storage.md +++ b/docs/architecture/storage.md @@ -521,8 +521,10 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр | `20-ods-tables.sql` | типизированное событие, версии заказа и обе таблицы ошибок | | `30-ods-views.sql` | актуальные события, текущие заказы и матвью разбора в ODS | | `40-stg-views.sql` | матвью приёма: чтец в сырьё | -| `45-dds-tables.sql` | физические модели заказа и сессии, доноры их партиций | -| `50-dds-views.sql` | публичные договоры события, заказа и сессии на языке модели DDS | +| `45-dds-tables.sql` | физические модели заказа, сессии и карты идентичностей, доноры партиций | +| `50-dds-views.sql` | публичные договоры сущностей DDS | +| `55-dm-tables.sql` | физические витрины выручки и трафика, донор партиций выручки | +| `60-dm-views.sql` | публичные договоры витрин DM | Справочники идут сразу за базами: зона `dic` не зависит ни от одного слоя, и правила порядка слоёв к ней не применяются. @@ -589,6 +591,11 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр | `dds` | `dds.session_v` | публичный договор сессии | | `dds` | `dds.identity_map_rep` / `_dist` | известная пара куки с пользователем; таблицы полного пересчёта | | `dds` | `dds.identity_map_v` | точный публичный договор карты идентичностей с `FINAL` | +| `dm` | `dm.revenue_daily_rep` / `_dist` | дневная выручка по категориям | +| `dm` | `dm.revenue_daily_stage_rep` / `_dist` | таблица-двойник, донор дневных партиций выручки | +| `dm` | `dm.revenue_daily_v` | публичный договор дневной выручки | +| `dm` | `dm.daily_traffic_rep` / `_dist` | версии дневных счетчиков трафика | +| `dm` | `dm.daily_traffic_v` | точные текущие счетчики трафика с `FINAL` | | `dic` | `dic.products_file` | чтец CSV-каталога, общего с генератором | | `dic` | `dic.products` | словарь товаров поверх подложки | @@ -598,9 +605,8 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр В слое DDS построены событие, заказ, сессия и карта идентичностей. Событие остаётся представлением над ODS; заказ и сессия гасят повтор заменой партиции; -карта пишет полный набор пар в версионную таблицу. Слой DM принесут следующие -тикеты этапа. Его состав задан разделом 7 мастер-спеки и переносится сюда по -мере постройки. +карта пишет полный набор пар в версионную таблицу. В DM выручка заменяет +партиции изменяемых дней, а трафик пишет полную историю новой версией. ## Что проверено diff --git a/docs/specs/2026-07-30-stand-v2-realism.md b/docs/specs/2026-07-30-stand-v2-realism.md index 79f5f3a..d6fbaef 100644 --- a/docs/specs/2026-07-30-stand-v2-realism.md +++ b/docs/specs/2026-07-30-stand-v2-realism.md @@ -367,8 +367,10 @@ CSV в репозитории (`data/catalog/products.csv`: `sku`, `name`, `cate ошибка в `*_errors`, а не молчаливые нули: имена CamelCase регистрозависимы, опечатка иначе не падает. - **Политика соединений**: по ключу ко-локации — обычное соединение с - комментарием, почему локальный результат корректен; по любому другому ключу - — только явный GLOBAL; `NOT IN` — только `GLOBAL NOT IN`. Сверка + комментарием, почему локальный результат корректен; если публичные + представления скрывают ко-локацию — решение записывается рядом с витриной. + По любому другому ключу — только явный GLOBAL; `NOT IN` — только + `GLOBAL NOT IN`. Сверка `purchase`↔заказ — легитимная GLOBAL-витрина (заказы малы). - **Конвейер без TRUNCATE**: поток версий — в ReplacingMergeTree, точное чтение @@ -505,8 +507,9 @@ Kafka день переигрывается генератором заново: счётчики `purchases`/`add_to_carts` оживают из таксономии, добавляется `declared_revenue` по UTM. - **`daily_traffic_v`** — расширяется парой «посетители» / «известные - пользователи» (обогащение через `dds.identity_map_v`, локальное соединение - по ключу ко-локации). + пользователи». Обогащение идет явным `GLOBAL LEFT JOIN` через + `dds.identity_map_v`: опыт на ClickHouse 26.3 показал, что публичные `_v` + скрывают ко-локацию от локальной подстановки. - `events_enriched_v`, `top_pages_daily_v`, `session_overview_v`, `dq_errors_daily_v` — переезжают на новую модель без смены роли: источник — `dds.event_v`, не `ods.event`. @@ -684,10 +687,6 @@ Kafka Engine на двух нодах снят с этого списка при `ReplacingMergeTree`. Проверять то, что заведомо случается и заведомо обезврежено, — работа без ответа на конце. -- Точную форму локального обогащения `daily_traffic_v` через публичные `_v` — - эмпирически при проектировании витрин #135. Общее поведение соединения двух - Distributed-таблиц уже проверено в - [исследовании](../research/2026-08-22-distributed-join.md). - Размер артефакта эталонного мира после пересборки. - Спорные API (Airflow Datasets/сенсоры, ClickHouse DDL) — перед кодом сверять через MCP Context7 (правило AGENTS.md). diff --git a/sql/ddl/00-databases.sql b/sql/ddl/00-databases.sql index 1433fc6..9f92b29 100644 --- a/sql/ddl/00-databases.sql +++ b/sql/ddl/00-databases.sql @@ -17,6 +17,9 @@ CREATE DATABASE IF NOT EXISTS ods ON CLUSTER clickstream_cluster; -- База DDS заводится здесь же по той же причине, что и ODS. CREATE DATABASE IF NOT EXISTS dds ON CLUSTER clickstream_cluster; +-- База витрин завершает цепочку STG → ODS → DDS → DM. +CREATE DATABASE IF NOT EXISTS dm ON CLUSTER clickstream_cluster; + -- Справочники стоят вне цепочки STG → ODS → DDS → DM: в хранилище их никто не -- производит, а читают их несколько слоёв — ADR 0012. Порядок слоёв к зоне не -- применяется, её файл идёт сразу за этим. diff --git a/sql/ddl/55-dm-tables.sql b/sql/ddl/55-dm-tables.sql new file mode 100644 index 0000000..6ccd51a --- /dev/null +++ b/sql/ddl/55-dm-tables.sql @@ -0,0 +1,78 @@ +-- DM: физические дневные витрины и донор партиций выручки. + +-- Выручка хранится в зерне «день заказа × категория товара». День служит +-- единицей пересборки, поэтому совпадает с ключом партиции. +CREATE TABLE IF NOT EXISTS dm.revenue_daily_rep ON CLUSTER clickstream_cluster +( + report_date Date, + product_category LowCardinality(String), + orders UInt64, + units UInt64, + revenue Decimal(18, 2), + aov Decimal(18, 2), + _load_id String, + _load_ts DateTime64(3, 'UTC') +) +ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/{table}', '{replica}') +PARTITION BY report_date +ORDER BY (report_date, product_category); + +-- Категория — единственное измерение кроме дня. Шесть категорий стартового +-- мира близки по объему, поэтому простой ключ раскладывает строки достаточно +-- ровно и не добавляет в учебный пример составной хеш без читателя. +CREATE TABLE IF NOT EXISTS dm.revenue_daily_dist ON CLUSTER clickstream_cluster +AS dm.revenue_daily_rep +ENGINE = Distributed( + 'clickstream_cluster', + 'dm', + 'revenue_daily_rep', + cityHash64(product_category) +); + +-- Донор объявлен через AS и шардируется тем же ключом, что цель. REPLACE +-- PARTITION идет по локальным таблицам, поэтому структура и раскладка донора +-- обязаны совпадать с целью (docs/architecture/storage.md, «Раскладка по +-- шардам»). +CREATE TABLE IF NOT EXISTS dm.revenue_daily_stage_rep ON CLUSTER clickstream_cluster +AS dm.revenue_daily_rep +ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/{table}', '{replica}') +PARTITION BY report_date +ORDER BY (report_date, product_category); + +CREATE TABLE IF NOT EXISTS dm.revenue_daily_stage_dist ON CLUSTER clickstream_cluster +AS dm.revenue_daily_stage_rep +ENGINE = Distributed( + 'clickstream_cluster', + 'dm', + 'revenue_daily_stage_rep', + cityHash64(product_category) +); + +-- Трафик пересчитывается целиком: новая связь куки с пользователем меняет +-- прошлые дни. Версия запуска позволяет точному представлению скрыть строки +-- прежних пересчетов без TRUNCATE и без партиций. +CREATE TABLE IF NOT EXISTS dm.daily_traffic_rep ON CLUSTER clickstream_cluster +( + report_date Date, + visitors UInt64, + known_users UInt64, + _load_id String, + _load_ts DateTime64(3, 'UTC') +) +ENGINE = ReplicatedReplacingMergeTree( + '/clickhouse/tables/{shard}/{database}/{table}', + '{replica}', + _load_ts +) +ORDER BY report_date; + +-- Все версии одного дня должны попадать на один шард, иначе FINAL не сможет +-- выбрать последнюю через границу шардов. +CREATE TABLE IF NOT EXISTS dm.daily_traffic_dist ON CLUSTER clickstream_cluster +AS dm.daily_traffic_rep +ENGINE = Distributed( + 'clickstream_cluster', + 'dm', + 'daily_traffic_rep', + cityHash64(report_date) +); diff --git a/sql/ddl/60-dm-views.sql b/sql/ddl/60-dm-views.sql new file mode 100644 index 0000000..4b132c2 --- /dev/null +++ b/sql/ddl/60-dm-views.sql @@ -0,0 +1,28 @@ +-- DM: публичные договоры стендовых витрин. + +-- День и категория — зерно выручки. Служебные колонки показывают запуск, +-- который последним заменил дневную партицию. +CREATE VIEW IF NOT EXISTS dm.revenue_daily_v ON CLUSTER clickstream_cluster +AS +SELECT + report_date, + product_category, + orders, + units, + revenue, + aov, + _load_id, + _load_ts +FROM dm.revenue_daily_dist; + +-- Полный пересчет пишет новую версию каждого дня. FINAL оставляет текущие +-- счетчики и прячет повторы запусков от потребителя. +CREATE VIEW IF NOT EXISTS dm.daily_traffic_v ON CLUSTER clickstream_cluster +AS +SELECT + report_date, + visitors, + known_users, + _load_id, + _load_ts +FROM dm.daily_traffic_dist FINAL; diff --git a/sql/dm/daily_traffic_rebuild.sql b/sql/dm/daily_traffic_rebuild.sql new file mode 100644 index 0000000..02318fc --- /dev/null +++ b/sql/dm/daily_traffic_rebuild.sql @@ -0,0 +1,27 @@ +-- DM: полный пересчет дневного трафика. +-- +-- Публичные _v скрывают Distributed-таблицы от локальной подстановки: +-- EXPLAIN на ClickHouse 26.3 показал чтение обоих шардов на инициаторе. +-- Поэтому GLOBAL назван явно, а физические таблицы не подставлены. LEFT JOIN +-- сохраняет анонимные сессии. +INSERT INTO dm.daily_traffic_dist +( + report_date, + visitors, + known_users, + _load_id, + _load_ts +) +SELECT + sessions.session_date, + uniqExact(sessions.client_id), + uniqExactIf(identities.user_id, identities.user_id != 0), + {load_id:String}, + now64(3, 'UTC') +FROM dds.session_v AS sessions +GLOBAL LEFT JOIN dds.identity_map_v AS identities USING (client_id) +GROUP BY sessions.session_date +-- Ноль оставляет итоговую группировку на инициаторе: иначе каждый шард записал +-- бы свою строку дня, а FINAL через границу шардов их не объединяет. +SETTINGS distributed_foreground_insert = 1, + parallel_distributed_insert_select = 0; diff --git a/sql/dm/revenue_daily_rebuild.sql b/sql/dm/revenue_daily_rebuild.sql new file mode 100644 index 0000000..46f4e24 --- /dev/null +++ b/sql/dm/revenue_daily_rebuild.sql @@ -0,0 +1,55 @@ +-- DM: сборка выручки в донор перед заменой дневных партиций. +-- DROP PARTITION ALL очищает локальные таблицы на каждом узле; причина не +-- применять TRUNCATE разобрана в соседнем sql/dds/order_rebuild.sql. + +ALTER TABLE dm.revenue_daily_stage_rep ON CLUSTER clickstream_cluster +DROP PARTITION ALL; + +INSERT INTO dm.revenue_daily_stage_dist +( + report_date, + product_category, + orders, + units, + revenue, + aov, + _load_id, + _load_ts +) +SELECT + report_date, + product_category, + orders, + units, + CAST(revenue, 'Decimal(18, 2)'), + -- Обычное деление Decimal усекает доли копейки вместо округления; + -- запас масштаба сохраняет их до денежного результата. + CAST( + round(divideDecimal(revenue, toDecimal128(orders, 0), 6), 2), + 'Decimal(18, 2)' + ), + {load_id:String}, + now64(3, 'UTC') +FROM +( + SELECT + order_date AS report_date, + dictGet('dic.products', 'category', item_sku) AS product_category, + uniqExact(order_id) AS orders, + sum(item_quantity) AS units, + sum(item_price * item_quantity) AS revenue + FROM dds.order_v + ARRAY JOIN + item_sku, + item_quantity, + item_price + WHERE status = 'paid' + AND order_date BETWEEN {first_day:Date} AND {last_day:Date} + GROUP BY + report_date, + product_category +) +-- Источник лежит по заказу, цель — по категории. Ноль заставляет инициатор +-- слить частичные агрегаты и разложить готовые строки по ключу цели. +SETTINGS distributed_foreground_insert = 1, + parallel_distributed_insert_select = 0; diff --git a/sql/dm/revenue_daily_replace.sql b/sql/dm/revenue_daily_replace.sql new file mode 100644 index 0000000..0fedd96 --- /dev/null +++ b/sql/dm/revenue_daily_replace.sql @@ -0,0 +1,8 @@ +-- DM: подмена одного дня выручки готовой партицией донора. +-- +-- День приходит из результата revenue_daily_scope.sql, а не от пользователя. +-- Jinja нужна потому, что ClickHouse не подставляет параметр в PARTITION. +-- Замена идет на локальной таблице ON CLUSTER: граница атомарности и повтор +-- разобраны на таком же шаге в sql/dds/order_replace.sql. +ALTER TABLE dm.revenue_daily_rep ON CLUSTER clickstream_cluster +REPLACE PARTITION '{{ params.day }}' FROM dm.revenue_daily_stage_rep; diff --git a/sql/dm/revenue_daily_scope.sql b/sql/dm/revenue_daily_scope.sql new file mode 100644 index 0000000..2eea500 --- /dev/null +++ b/sql/dm/revenue_daily_scope.sql @@ -0,0 +1,35 @@ +-- DM: объем пересборки выручки. +-- +-- Последний слепок ODS называет дни, состояние которых источник еще может +-- изменить. Сам агрегат читает только DDS; ODS здесь нужен лишь для границ. +-- Первый запуск и повтор после частичного отказа расширяют левую границу до +-- первого дня оплаченных заказов, которого еще нет в витрине. +-- +-- Файл кончается без точки с запятой: даг читает его одним запросом и ждет +-- одну строку с границами. +WITH + (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 minOrNull(report_date) + FROM (SELECT DISTINCT order_date AS report_date FROM dds.order_v WHERE status = 'paid') + -- Правая сторона тоже распределенная: GLOBAL собирает множество дней + -- на инициаторе и не превращает запрос в запрещенное двойное чтение. + WHERE report_date GLOBAL NOT IN ( + SELECT report_date FROM dm.revenue_daily_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 dds.order_v) > 0 + AND (SELECT count() FROM ods.order_dist) > 0 -- 2.54.0 From 741820789f4efc84bc52e40ec949580ef8fef191 Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Wed, 26 Aug 2026 14:56:33 -0400 Subject: [PATCH 2/2] =?UTF-8?q?refactor(dm):=20=D1=83=D1=82=D0=BE=D1=87?= =?UTF-8?q?=D0=BD=D0=B5=D0=BD=D1=8B=20=D0=BE=D0=B3=D1=80=D0=B0=D0=BD=D0=B8?= =?UTF-8?q?=D1=87=D0=B5=D0=BD=D0=B8=D1=8F=20DAG=20=D0=B8=20=D0=BF=D0=BE?= =?UTF-8?q?=D1=8F=D1=81=D0=BD=D0=B5=D0=BD=D0=B8=D1=8F=20SQL?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Зачем: - Ревью выявило неясные причины ограничений и защитных условий в учебном образце. - Что: - Спекулятивный общий потолок задач заменен адресным лимитом размноженной замены с точным описанием OOM. - Объяснены нулевое значение непарного LEFT JOIN и опасность пустого DDS при замене партиций. - Добавлена ссылка на разбор области пересборки, а общий штамп про Context7 удален. - Проверка: - make lint, make config-test, make smoke; успешный запуск dm_transform. --- dags/dm_transform.py | 16 +++++++--------- sql/dm/daily_traffic_rebuild.sql | 2 ++ sql/dm/revenue_daily_scope.sql | 4 ++++ 3 files changed, 13 insertions(+), 9 deletions(-) diff --git a/dags/dm_transform.py b/dags/dm_transform.py index 066b73f..4915ef5 100644 --- a/dags/dm_transform.py +++ b/dags/dm_transform.py @@ -3,9 +3,6 @@ Выручка наследует ритм заказов и заменяет дневные партиции изменяемого окна. Трафик наследует ретроспективность карты идентичностей и пересчитывает всю историю новой версией. Объем каждого прогона выводится из состояния слоев. - -Форма DAG и динамическая замена сверены с публичным API Airflow 3 через -Context7; DDL и REPLACE PARTITION — с документацией ClickHouse. """ from __future__ import annotations @@ -26,7 +23,7 @@ CLICKHOUSE_CONNECTION = "clickhouse_default" def _days(first_day: datetime.date, last_day: datetime.date) -> list[str]: - """Дни замены строками от левой до правой границы включительно.""" + """Дни отрезка строками - в таком виде их ждет замена партиции.""" return [ (first_day + datetime.timedelta(days=shift)).isoformat() for shift in range((last_day - first_day).days + 1) @@ -49,7 +46,6 @@ def replacements(scope: dict[str, object]) -> list[dict[str, object]]: start_date=START_DATE, is_paused_upon_creation=False, max_active_runs=1, - max_active_tasks=3, template_searchpath=str(SQL_ROOT), tags=["dm"], ) @@ -70,7 +66,8 @@ def dm_transform(): row = hook.get_first(query) if row is None: raise AirflowException( - "нет слепка заказов в ODS или готовых заказов в DDS" + "нет слепка заказов в ODS или готовых заказов в DDS; " + "дождитесь загрузки слепка и сначала выполните dds_transform" ) first_day, last_day = row days = _days(first_day, last_day) @@ -106,9 +103,10 @@ def dm_transform(): conn_id=CLICKHOUSE_CONNECTION, sql="dm/revenue_daily_replace.sql", do_xcom_push=False, - # Каждая замена поднимает отдельный процесс работника. Три - # одновременные задачи выбили его по памяти на первом прогоне; - # самим дням параллельность ничего не дает. + # Airflow 3 применяет этот лимит и к размноженным экземплярам + # (сверено через Context7). Каждая замена - отдельный процесс + # LocalExecutor; три разом исчерпали память контейнера scheduler, + # а самим дням параллельность ничего не дает. max_active_tis_per_dag=1, ).expand_kwargs(replacements(days)) diff --git a/sql/dm/daily_traffic_rebuild.sql b/sql/dm/daily_traffic_rebuild.sql index 02318fc..83d3e1c 100644 --- a/sql/dm/daily_traffic_rebuild.sql +++ b/sql/dm/daily_traffic_rebuild.sql @@ -15,6 +15,8 @@ INSERT INTO dm.daily_traffic_dist SELECT sessions.session_date, uniqExact(sessions.client_id), + -- При стендовом join_use_nulls=0 непарный UInt64 получает 0. Условие + -- не дает анонимной сессии превратиться в известного пользователя. uniqExactIf(identities.user_id, identities.user_id != 0), {load_id:String}, now64(3, 'UTC') diff --git a/sql/dm/revenue_daily_scope.sql b/sql/dm/revenue_daily_scope.sql index 2eea500..c15154f 100644 --- a/sql/dm/revenue_daily_scope.sql +++ b/sql/dm/revenue_daily_scope.sql @@ -4,6 +4,8 @@ -- изменить. Сам агрегат читает только DDS; ODS здесь нужен лишь для границ. -- Первый запуск и повтор после частичного отказа расширяют левую границу до -- первого дня оплаченных заказов, которого еще нет в витрине. +-- Почему хвост начинается с первого пропуска и чем за это платит ODS, +-- разобрано в sql/dds/order_scope.sql. -- -- Файл кончается без точки с запятой: даг читает его одним запросом и ждет -- одну строку с границами. @@ -31,5 +33,7 @@ WITH SELECT least(window_start, coalesce(first_missing_day, window_start)) AS first_day, window_end AS last_day +-- Без ODS неизвестно окно. Без готового DDS донор останется пустым, а замена +-- сотрет уже собранные дни выручки. Поэтому для прогона нужны оба источника. WHERE (SELECT count() FROM dds.order_v) > 0 AND (SELECT count() FROM ods.order_dist) > 0 -- 2.54.0