From 1e8b83ffbb27d59b9f3577ae87ee9079f4ded306 Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Wed, 26 Aug 2026 05:20:39 -0400 Subject: [PATCH 1/2] =?UTF-8?q?feat(dds):=20=D0=B4=D0=BE=D0=B1=D0=B0=D0=B2?= =?UTF-8?q?=D0=BB=D0=B5=D0=BD=D0=B0=20=D0=BC=D0=BE=D0=B4=D0=B5=D0=BB=D1=8C?= =?UTF-8?q?=20=D0=B7=D0=B0=D0=BA=D0=B0=D0=B7=D0=B0=20=D0=B8=20=D0=BF=D0=B5?= =?UTF-8?q?=D1=80=D0=B5=D1=81=D0=B1=D0=BE=D1=80=D0=BA=D0=B0=20=D0=BE=D0=BA?= =?UTF-8?q?=D0=BD=D0=B0=20=D0=B7=D0=B0=D0=BC=D0=B5=D0=BD=D0=BE=D0=B9=20?= =?UTF-8?q?=D0=BF=D0=B0=D1=80=D1=82=D0=B8=D1=86=D0=B8=D0=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Зачем: - Дать слою первую физическую сущность и показать идемпотентность, устроенную иначе, чем в приёме: заменой партиции, а не колонкой версии. - Родить механику дневной пересборки на кластере — образец для сессий и витрин. Что: - Добавлены dds.order_rep/_dist, двойник-донор dds.order_stage_rep/_dist и публичный договор dds.order_v с разбором позиций в три массива. - Добавлены запросы объёма прогона, сборки в двойник и замены партиции по дню. - Добавлен даг dds_transform: группа задач order, замена — размноженной задачей. - Окно изменяемости K пришло на сторону стенда якорем compose; владелец числа остаётся у генератора. - Объём считается от позднейшего дня заказа и первого недостающего дня: ярлыки доставки ODS для этого непригодны, а максимум собранного дня прячет дыры. - Записано, чего конвейер о себе не знает, и адресовано контуру DQ. Проверка: - make lint - make smoke, make check-clickhouse, make check-services - Прогон по стартовому миру: 1917 заказов, счёт равен ods.order_v, каждый заказ на шарде своего cityHash64(order_id), сумма позиций равна items_total. - Повтор без нового слепка: содержимое то же, _load_id новый; после слепка пересобрано окно, день за окном не тронут. - Снесённая руками партиция 2026-06-01 восстановлена побайтово. Co-Authored-By: Claude Opus 5 (1M context) --- compose.yaml | 11 +++ dags/dds_transform.py | 142 +++++++++++++++++++++++++++++++++++ docs/architecture/etl.md | 45 ++++++++++- docs/architecture/storage.md | 52 +++++++++++-- sql/ddl/45-dds-tables.sql | 95 +++++++++++++++++++++++ sql/ddl/50-dds-views.sql | 34 +++++++++ sql/dds/order_rebuild.sql | 89 ++++++++++++++++++++++ sql/dds/order_replace.sql | 28 +++++++ sql/dds/order_scope.sql | 65 ++++++++++++++++ 9 files changed, 555 insertions(+), 6 deletions(-) create mode 100644 dags/dds_transform.py create mode 100644 sql/ddl/45-dds-tables.sql create mode 100644 sql/dds/order_rebuild.sql create mode 100644 sql/dds/order_replace.sql create mode 100644 sql/dds/order_scope.sql diff --git a/compose.yaml b/compose.yaml index 56671e3..70f2129 100644 --- a/compose.yaml +++ b/compose.yaml @@ -48,6 +48,16 @@ x-kafka-target: &kafka-target x-world-starting-days: &world-starting-days "8" +# Окно изменяемости заказа: слепок бэкенда несёт заказы последних K дней, за +# окном заказ замерзает. Число принадлежит модельному миру, и владелец у него +# один — generator/world.py (ORDER_WINDOW_DAYS): сделать окно настройкой +# генератора значило бы разрешить менять физику мира снаружи, а опись мира +# держится на её неизменности. Стенду копия нужна затем, что по этому же числу +# трансформация DDS решает, какие дни пересобирать. Разойдутся — пересборка +# промахнётся по краю окна, и заметит это не проверка, а менти со вчерашней +# выручкой. +x-order-window-days: &order-window-days "7" + x-airflow-common: &airflow-common image: clickstream-airflow:local build: @@ -77,6 +87,7 @@ x-airflow-common: &airflow-common STAND_NETWORK: ${COMPOSE_PROJECT_NAME}_default WORLD_STARTING_DAYS: *world-starting-days KAFKA_ORDERS_TOPIC: *orders-topic + ORDER_WINDOW_DAYS: *order-window-days volumes: - ./dags:/opt/airflow/dags:ro # Запросы дагов: даг называет файл, а обработчик DAG при разборе загружает diff --git a/dags/dds_transform.py b/dags/dds_transform.py new file mode 100644 index 0000000..39b854a --- /dev/null +++ b/dags/dds_transform.py @@ -0,0 +1,142 @@ +"""Трансформации слоя DDS: пересборка дней, которые ещё могут измениться. + +Даг собирает модель поверх принятого ODS. Первая его сущность — заказ; сессии +и карта идентичностей встанут соседними группами задач. + +Расписания у дага нет и параметров тоже: объём каждого прогона он выводит сам +из состояния слоёв, поэтому один и тот же запуск одинаково верен из триггера +дневного конвейера, с кнопки менти и после падения +(docs/architecture/etl.md, «Объём прогона»). + +Идемпотентность здесь устроена не так, как в приёме заказов, и в этом весь +урок сущности. Приём гасит повтор колонкой версии: ReplacingMergeTree +оставляет позднюю строку. Трансформация версий не знает — она собирает день +заново в таблицу-двойник и атомарно подменяет им партицию цели. Поэтому +повтор прогона не меняет делового содержимого слоя, а строки пересобранных +дней получают новый _load_id: видно, каким запуском собран каждый день. + +Замена партиции идёт по локальным таблицам с ON CLUSTER: по Distributed +операции с партициями отвечают кодом 48 (docs/architecture/storage.md, «Что +проверено»). Форма двойника и цена ошибки в ключе шардирования разобраны в +sql/ddl/45-dds-tables.sql. +""" + +from __future__ import annotations + +import datetime +import logging +import os +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" + +# Окно изменяемости заказа, модельные дни. Число мира, а не настройка стенда: +# владелец — generator/world.py (ORDER_WINDOW_DAYS), который этим окном +# собирает слепок. Здесь оно читается, чтобы хранилище знало, какие дни ещё +# могут измениться. Разойдутся два объявления — пересборка промахнётся по краю +# окна, поэтому оба названы в compose.yaml рядом и с этой оговоркой. +ORDER_WINDOW_DAYS = int(os.environ["ORDER_WINDOW_DAYS"]) + + +@dag( + dag_id="dds_transform", + schedule=None, + start_date=START_DATE, + is_paused_upon_creation=False, + # Два прогона разом собирали бы один и тот же двойник: второй затёр бы + # чужую сборку между вставкой и заменой партиций. + max_active_runs=1, + template_searchpath=str(SQL_ROOT), + tags=["dds"], +) +def dds_transform(): + """Пересобрать сущности DDS по текущему состоянию источников.""" + + @task_group(group_id="order") + def order(): + """Заказ: дышащее окно плюс дни, которых в слое ещё нет.""" + + @task + def scope() -> dict[str, object]: + """Спросить у хранилища, какие дни собирает этот запуск.""" + query = (SQL_ROOT / "dds" / "order_scope.sql").read_text(encoding="utf-8") + hook = ClickHouseHook(clickhouse_conn_id=CLICKHOUSE_CONNECTION) + row = hook.get_first(query, parameters={"window_days": ORDER_WINDOW_DAYS}) + # Пустой ответ означает пустой источник: собирать нечего, и + # зеленеть на этом нельзя — прогон встаёт с внятной причиной. + if row is None: + raise AirflowException( + "в ods.order нет ни одного заказа: собирать нечего; " + "проверьте, дошёл ли слепок до хранилища" + ) + first_day, last_day = row + days = [ + (first_day + datetime.timedelta(days=shift)).isoformat() + for shift in range((last_day - first_day).days + 1) + ] + logging.info( + "объём пересборки: %s … %s, дней %d", + first_day, + last_day, + len(days), + ) + return { + "first_day": first_day.isoformat(), + "last_day": last_day.isoformat(), + "days": days, + } + + @task + def replacements(scope: dict[str, object]) -> list[dict[str, object]]: + """Разложить дни в аргументы размноженной задачи замены. + + Шаг существует ради ограничения Airflow: размножать задачу можно + только по целому возврату другой задачи — ни по ключу словаря, ни + по карте поверх него. Отсюда отдельный возврат под один список. + + День едет в запрос подстановкой Jinja, а не параметром ClickHouse: + в ALTER … PARTITION сервер параметр не подставляет — замерено, + отвечает синтаксической ошибкой. Значение при этом не внешнее, его + вернул сам ClickHouse запросом объёма. + """ + return [{"params": {"day": day}} for day in scope["days"]] + + days = scope() + + rebuild = SQLExecuteQueryOperator( + task_id="rebuild", + conn_id=CLICKHOUSE_CONNECTION, + sql="dds/order_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="dds/order_replace.sql", + do_xcom_push=False, + ).expand_kwargs(replacements(days)) + + rebuild >> replace + + order() + + +dds_transform() diff --git a/docs/architecture/etl.md b/docs/architecture/etl.md index 4ce9370..1d57ec1 100644 --- a/docs/architecture/etl.md +++ b/docs/architecture/etl.md @@ -41,12 +41,30 @@ Параметров у конвейера нет: объём он выводит сам из `world_position` и состояния слоёв. Сессии догоняют хвост — от последнего собранного дня до -последнего прожитого; заказы пересобирают дышащее окно K от позиции; карта +последнего прожитого; заказы пересобирают дышащее окно K и тот же хвост; карта идентичностей пересчитывается целиком; правила витрин — у [слоя DM](dm.md). Поэтому один и тот же запуск одинаково верен из триггера, с кнопки менти и после падения, а разгону на семь дней хватает одного прогона. +Окно заказов отсчитывается от **последнего дня, о котором рассказал сам +источник**, — от позднейшего дня заказа в ODS. Рядом напрашиваются ещё два +способа, и оба отвергнуты по одной причине: они отвечают не на тот вопрос. + +Позиция мира говорит, какой день прожит, а не какой доехал: прогон дня D везёт +слепок дня D − 1, да и позиция ставится раньше трансформации, так что после +неудачного приёма она обгоняет хранилище. Слой обязан считать объём по тому, +что у него есть, а не по тому, что мир о себе заявляет. + +Ярлыки доставки — `snapshot_date`, `_load_id`, `_load_ts` — не отвечают тоже, и +это свойство ODS, а не недосмотр: слой схлопывает доставки, сохраняя деловые +версии, поэтому у заказа, который не менялся, ярлык выжившей строки достаётся +жребием ([приём заказов](orders/ingestion.md), «Роль ODS»). Замер на стартовом +мире: последний слепок нёс все 1662 заказа, а его ярлык сохранили 1188 — +отбор «дни, у которых `snapshot_date` равен последнему» потерял бы четверть +заказов и молча пропустил бы дальний край окна, где заказы давно замёрзли. +К времени создания это не относится: у всех версий заказа оно одно. + ## Готовность источника Первый шаг `dds_transform` — дождаться, пока хвост топика `hits` дожёван: @@ -67,6 +85,31 @@ замены: полная перезапись в `ReplacingMergeTree` — так живут карта идентичностей и витрина `daily_traffic` ([слой DM](dm.md)). +## Чего конвейер о себе не знает + +Трансформация собирает слой по правилу, а не сверяет его с источником. Отсюда +четыре вещи, которых она не заметит, и у всех четырёх один адрес — контур DQ. + +- **Окно K разъехалось с источником.** Число живёт двумя копиями: у генератора + в `world.py` и у стенда в `compose.yaml`. Разойдутся — слой начнёт + недобирать край окна или переписывать замёрзший день, и оба прогона будут + зелёными. +- **Пустой свежий день сдвигает край окна.** Если в позднейший день источника + не родилось ни одного заказа, окно съезжает на день назад + ([`sql/dds/order_scope.sql`](../../sql/dds/order_scope.sql)). +- **Замена партиции прошла на одном шарде из двух.** День в слое есть — значит, + объём следующего прогона его не тронет, а слой останется собранным наполовину + из старого, наполовину из нового. +- **Брак внутри `items` становится нулями.** Строгий приём внутрь позиций не + смотрит, разбор по типу молчит, и цена без числа доедет до витрины выручки + нулём ([`sql/dds/order_rebuild.sql`](../../sql/dds/order_rebuild.sql)). + +Общее у них одно, и оно же подсказывает лекарство: ответить может только +сверка слоя с источником по дню — счёт заказов и позднейший `updated_at` с +обеих сторон. Трансформации такую сверку не поручают намеренно: она собирает, +а проверяет другой. Так две роли остаются различимы, и ни одна не +притворяется другой. + ## Не входит - Партиционная переобработка — переиграть произвольный прошлый день или diff --git a/docs/architecture/storage.md b/docs/architecture/storage.md index 8b64a17..c21d7b1 100644 --- a/docs/architecture/storage.md +++ b/docs/architecture/storage.md @@ -15,7 +15,9 @@ keeper, Kafka, каркас сервисов. Этап 2 собран: в `sql/d Тот же этап принёс первый справочник — словарь товаров из общего с генератором CSV-файла. Живёт он вне цепочки слоёв, в зоне `dic`, и читает не файл, а подложку над ним ([ADR 0012](../adr/0012-dictionary-home.md)). Этап 4 начат -публичным договором события `dds.event_v` над точным состоянием ODS. +публичным договором события `dds.event_v` над точным состоянием ODS, а следом +получил первую физическую модель слоя — заказ: его пересобирает заменой +партиций даг `dds_transform`. Дальше по тексту устройство описано так, как оно проектируется; построенное от заложенного отличает карта таблиц в конце. @@ -129,6 +131,14 @@ FROM источник_dist`. Заводское значение `parallel_distr разложена по шардам по тому же ключу. Форму донора следует решать вместе с таким объектом, а не переносить на ODS заранее. +Первым такой донор понадобился заказу DDS, и правило на нём видно целиком: +`dds.order_stage_rep` объявлен через `AS dds.order_rep` и шардируется тем же +`cityHash64(order_id)`. Совпадение структуры требует сама операция, а совпадение +ключа — устройство замены: она идёт по локальным таблицам, шард за шардом, и +строка, попавшая в доноре не на свой шард, при подмене пропала бы с одного шарда +и задвоилась на другом. Доводы целиком — в +[`sql/ddl/45-dds-tables.sql`](../../sql/ddl/45-dds-tables.sql). + ## Служебные колонки Собственные колонки не повторяют имён виртуальных. Виртуальные даёт движок: @@ -508,7 +518,8 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр | `20-ods-tables.sql` | типизированное событие, версии заказа и обе таблицы ошибок | | `30-ods-views.sql` | актуальные события, текущие заказы и матвью разбора в ODS | | `40-stg-views.sql` | матвью приёма: чтец в сырьё | -| `50-dds-views.sql` | публичный договор события на языке модели DDS | +| `45-dds-tables.sql` | физическая модель заказа и таблица-двойник, донор её партиций | +| `50-dds-views.sql` | публичные договоры события и заказа на языке модели DDS | Справочники идут сразу за базами: зона `dic` не зависит ни от одного слоя, и правила порядка слоёв к ней не применяются. @@ -567,6 +578,9 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр | `ods` | `ods.order_v` | текущая версия заказа на языке источника | | `ods` | `ods.order_errors_rep` / `_dist` | строки слепка, не прошедшие строгий приём | | `dds` | `dds.event_v` | событие на языке бизнес-модели без физической копии | +| `dds` | `dds.order_rep` / `_dist` | заказ в текущем состоянии, партиция — день заказа | +| `dds` | `dds.order_stage_rep` / `_dist` | двойник заказа: в нём день собирается перед заменой партиции | +| `dds` | `dds.order_v` | публичный договор заказа | | `dic` | `dic.products_file` | чтец CSV-каталога, общего с генератором | | `dic` | `dic.products` | словарь товаров поверх подложки | @@ -574,9 +588,10 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр `INSERT SELECT` одной задачи `parse_batch` из файла `sql/ods/order_parse.sql`. -Слой DDS начат представлением `dds.event_v`; остальные его сущности и слой DM -появятся в следующих тикетах этапа. Их состав задан разделом 7 мастер-спеки и -переносится сюда по мере постройки. +В слое DDS построены событие и заказ, и устроены они намеренно по-разному: +событие остаётся представлением над ODS, заказ получил физическую модель. +Сессии, карту идентичностей и слой DM принесут следующие тикеты этапа. Их +состав задан разделом 7 мастер-спеки и переносится сюда по мере постройки. ## Что проверено @@ -788,5 +803,32 @@ Kafka с пустым значением (ноль байт) и запись-н таблице. Это вычитано, а не измерено. На устройство приёма оговорка не влияет: служебные колонки мы заполняем выражением при любом ответе. +**Проверка замены партиций 26 августа 2026 года (#151).** Context7 подтвердил +договор `REPLACE PARTITION`: `ON CLUSTER` в синтаксисе есть, а источник и цель +обязаны совпадать структурой, ключом партиции, ключом сортировки, первичным +ключом и политикой хранения. На закреплённом ClickHouse 26.3 замена по локальным +таблицам с `ON CLUSTER` прошла на обеих нодах, а сборка стартового мира в +двойник разложила 1662 заказа строго по ключу: шард 1 получил только чётные +`cityHash64(order_id)`, шард 2 — только нечётные. + +Тем же днём измерены ещё три поведения ClickHouse. + +- **Параметр в `ALTER … PARTITION` не подставляется.** Запрос с `{day:Date}` в + выражении партиции отвечает кодом 62, синтаксической ошибкой; литерал + проходит. Отсюда единственное место в репозитории, где значение попадает в + запрос подстановкой Jinja, а не параметром ClickHouse + ([`sql/dds/order_replace.sql`](../../sql/dds/order_replace.sql)). +- **`TRUNCATE` — отдельное право, в `ALTER` оно не входит.** Роль `etl`, у + которой есть `ALTER ON dds.*`, получает на `TRUNCATE` код 497, а + `ALTER TABLE … DROP PARTITION ALL` на той же таблице проходит. Двойник + поэтому чистится снятием партиций: у операции нашлась форма, на которую право + уже выдано, и расширять доступ не понадобилось. +- **`JSONExtract` читает `Decimal` прямо из строки JSON.** Цена на проводе — + строка с двумя знаками (`"1289.99"`), и разбор в + `Array(Tuple(sku String, qty UInt32, price Decimal(18, 2)))` даёт число без + промежуточного `toDecimal64`. Проверено на живом слепке; честность разбора + сторожит инвариант бэкенда — сумма «цена × количество» совпала с `items_total` + у всех 1662 заказов. + **Сказано по памяти, проверки нет.** Группа пуста: оба утверждения, ждавшие матвью разбора, закрыты опытами при исполнении #43 и переехали выше. diff --git a/sql/ddl/45-dds-tables.sql b/sql/ddl/45-dds-tables.sql new file mode 100644 index 0000000..edc19ee --- /dev/null +++ b/sql/ddl/45-dds-tables.sql @@ -0,0 +1,95 @@ +-- DDS: физическая модель заказа и донор её партиций. +-- +-- Зерно — заказ в текущем состоянии, ключ order_id. Историю версий слой не +-- хранит: у неё нет читателя ни в одной витрине, а ODS её и не обещает +-- (docs/architecture/dds/order.md, «Зерно»). +-- +-- Идемпотентность здесь устроена иначе, чем в приёме, и это главный урок +-- сущности. В ODS повтор гасит колонка версии: ReplacingMergeTree(updated_at) +-- оставляет позднюю строку. Здесь колонки версии нет вовсе — повтор гасит +-- замена партиции целиком: день собирается заново и подменяется одним +-- атомарным движением. Поэтому каждая строка таблицы — ровно один заказ, а +-- чтению не нужен FINAL. + +-- Локальная таблица шарда. +-- +-- Партиция — order_date, день заказа в поясе счётчика. Партиция дневная не +-- ради размера куска, а потому что день — единица пересборки: пока заказ +-- внутри окна изменяемости, его день «дышит» и собирается заново, за окном +-- замерзает. Ключ партиции и есть контракт трансформации с хранилищем. +-- +-- Ключ сортировки — order_id: единственный устойчивый ключ сущности, он же +-- ключ шардирования ниже. Колонки версии в движке нет: см. шапку файла. +CREATE TABLE IF NOT EXISTS dds.order_rep ON CLUSTER clickstream_cluster +( + order_id String, + user_id UInt64, + status LowCardinality(String), + -- День заказа посчитан один раз при загрузке, по образцу EventDate у + -- события: суточные срезы группируются по готовой дате и пояса не + -- упоминают (docs/architecture/storage.md, «Часовые пояса»). + order_date Date, + -- Времена лежат местными: DDS обслуживает человека с дашбордом, и пояс + -- пересчитан один раз здесь, а не в каждом отчёте. Линза названа в типе, + -- поэтому пояс сервера на эти колонки не влияет. + created_at DateTime64(3, 'Europe/Samara'), + updated_at DateTime64(3, 'Europe/Samara'), + items_total Decimal(18, 2), + discount Decimal(18, 2), + delivery Decimal(18, 2), + total Decimal(18, 2), + -- Позиции — три массива одной длины: артикул, количество, цена. Родная + -- для колоночного хранилища денормализация: позиция не живёт без заказа и + -- своего ключа не имеет, а разворачивает массивы читатель — ARRAY JOIN в + -- витрине выручки. Форма записи выбрана по дому: широкое событие хранит + -- повторяющиеся группы Метрики такими же параллельными колонками + -- (productID, productPrice, productQuantity в ods.event_rep), а не Nested. + item_sku Array(String), + item_quantity Array(UInt32), + item_price Array(Decimal(18, 2)), + -- Координаты собственной загрузки слоя: каким запуском и когда собран + -- день. При замене партиции они одинаковы у всех строк дня. Ярлыки ODS + -- сюда не переезжают — почему, разобрано в доке модели. + _load_id String, + _load_ts DateTime64(3, 'UTC') +) +ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/{table}', '{replica}') +PARTITION BY order_date +ORDER BY order_id; + +-- Лицо загрузки и диагностики. Ключ шардирования — cityHash64(order_id), тот +-- же, что у ods.order_dist: заказ остаётся на одном шарде на всём пути через +-- слои, и соединения по зерну заказа обходятся без GLOBAL +-- (docs/architecture/storage.md, «Раскладка по шардам»). +CREATE TABLE IF NOT EXISTS dds.order_dist ON CLUSTER clickstream_cluster +AS dds.order_rep +ENGINE = Distributed('clickstream_cluster', 'dds', 'order_rep', cityHash64(order_id)); + +-- Донор партиций: таблица-двойник, в которой день собирается перед подменой. +-- +-- Двойник нужен потому, что REPLACE PARTITION берёт готовый кусок из другой +-- таблицы, а не из результата запроса. Собрать день прямо в цель нельзя: +-- вставка добавила бы строки к уже лежащим, и повтор прогона удвоил бы день. +-- +-- Форма двойника задана правилом донора: структура, ключ партиции, ключ +-- сортировки, первичный ключ и политика хранения обязаны совпадать с целью — +-- это требование самой операции (проверено по документации ClickHouse, +-- sql-reference/statements/alter/partition). Отсюда AS dds.order_rep вместо +-- повторённого списка колонок: расхождение становится невозможным по +-- построению и не держится на внимательности того, кто правит. +-- +-- Своё имя в keeper двойник получает от макроса {table}, поэтому две таблицы +-- не делят путь репликации. +CREATE TABLE IF NOT EXISTS dds.order_stage_rep ON CLUSTER clickstream_cluster +AS dds.order_rep +ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/{table}', '{replica}') +PARTITION BY order_date +ORDER BY order_id; + +-- Двойник шардируется тем же ключом, что цель, и это не украшение: REPLACE +-- PARTITION идёт по локальным таблицам, шард за шардом. Ляг строка заказа в +-- доноре на другой шард, чем её место в цели, — подмена перенесла бы её не +-- туда, и заказ пропал бы с одного шарда и задвоился на другом. +CREATE TABLE IF NOT EXISTS dds.order_stage_dist ON CLUSTER clickstream_cluster +AS dds.order_stage_rep +ENGINE = Distributed('clickstream_cluster', 'dds', 'order_stage_rep', cityHash64(order_id)); diff --git a/sql/ddl/50-dds-views.sql b/sql/ddl/50-dds-views.sql index d3136cd..dcfed0b 100644 --- a/sql/ddl/50-dds-views.sql +++ b/sql/ddl/50-dds-views.sql @@ -65,3 +65,37 @@ SELECT productQuantity AS product_quantity, productEventType AS product_event_type FROM ods.event_v; + +-- DDS: публичный договор заказа на языке бизнес-модели. +-- +-- В отличие от события, у заказа под представлением лежит физическая модель: +-- разбор позиций и пересчёт пояса стоят денег, и платить их при каждом чтении +-- значило бы собирать модель заново на лету (docs/architecture/dds/order.md, +-- «Отклонённые варианты»). Представление здесь ничего не вычисляет — оно +-- только называет колонки, которые слой обещает читателю. +-- +-- FINAL нет и не будет: строка на заказ одна по построению, повтор гасит +-- замена партиции, а не колонка версии. +-- +-- Служебные колонки перечислены наравне с деловыми — это координаты +-- собственной загрузки слоя, и расследование происшествия начинается с них: +-- по _load_id видно, какой запуск собрал день. +CREATE VIEW IF NOT EXISTS dds.order_v ON CLUSTER clickstream_cluster +AS +SELECT + order_id, + user_id, + status, + order_date, + created_at, + updated_at, + items_total, + discount, + delivery, + total, + item_sku, + item_quantity, + item_price, + _load_id, + _load_ts +FROM dds.order_dist; diff --git a/sql/dds/order_rebuild.sql b/sql/dds/order_rebuild.sql new file mode 100644 index 0000000..f886371 --- /dev/null +++ b/sql/dds/order_rebuild.sql @@ -0,0 +1,89 @@ +-- DDS: сборка дней заказа в таблицу-двойник перед заменой партиций. +-- +-- Двойник очищается целиком в начале сборки, а не в конце: собранное остаётся +-- лежать до следующего запуска, и разбор происшествия видит ровно то, чем +-- подменили день. Очистка обязана быть первой ещё и по делу — вставка поверх +-- прежних строк удвоила бы день в доноре, а REPLACE PARTITION перенёс бы +-- удвоение в цель. +-- +-- Дни собираются одной вставкой, а не по дню за раз: источник читается один +-- раз на весь отрезок. Кто раскладывает строки по шардам — вставка через +-- Distributed, ключ cityHash64(order_id) — разобрано в DDL двойника. +-- +-- Чистит двойник DROP PARTITION ALL, а не TRUNCATE, и это не вкусовщина: +-- TRUNCATE — отдельное право, роли etl оно не выдано, а снятие всех партиций +-- входит в ALTER, который у неё есть (ADR 0007). Замерено: TRUNCATE отвечает +-- ролью etl кодом 497, DROP PARTITION ALL проходит. Расширять права ради +-- операции, для которой уже есть разрешённая форма, значило бы платить +-- доступом за синоним. + +ALTER TABLE dds.order_stage_rep ON CLUSTER clickstream_cluster DROP PARTITION ALL; + +INSERT INTO dds.order_stage_dist +( + order_id, + user_id, + status, + order_date, + created_at, + updated_at, + items_total, + discount, + delivery, + total, + item_sku, + item_quantity, + item_price, + _load_id, + _load_ts +) +-- Позиции разбираются здесь один раз, на границе ODS → DDS. Цена на проводе — +-- строка с двумя знаками, и JSONExtract читает её прямо в Decimal: замерено на +-- живом слепке, промежуточный toDecimal64 не нужен. +-- +-- Разбор по типу молчалив, и знать об этом надо заранее. Строгий приём ODS +-- проверил только, что items — массив, внутрь он не смотрел; здесь тоже никто +-- не смотрит. Замерено: qty вида "oops" даёт 0, пропущенная цена даёт 0, +-- скаляр вместо объекта выпадает из массива целиком. На каноническом слепке +-- генератора такого не бывает, и сумма «цена × количество» сходится с +-- items_total у всех заказов, но сходится она по свойству источника, а не +-- потому, что её здесь кто-то проверяет. Постоянное место такой проверки — +-- контур DQ этапа 4; пока она делается руками при приёмке. +WITH JSONExtract( + items, 'Array(Tuple(sku String, qty UInt32, price Decimal(18, 2)))' + ) AS lines +SELECT + order_id, + user_id, + status, + -- День заказа и оба времени переезжают в пояс счётчика: DDS обслуживает + -- человека с дашбордом, и пояс называется здесь в последний раз. + toDate(toTimeZone(created_at, 'Europe/Samara')), + toTimeZone(created_at, 'Europe/Samara'), + toTimeZone(updated_at, 'Europe/Samara'), + items_total, + discount, + delivery, + total, + arrayMap(line -> line.sku, lines), + arrayMap(line -> line.qty, lines), + arrayMap(line -> line.price, lines), + -- Координаты запуска, собравшего день: у всех строк дня они одни. + {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, + -- поэтому день слева берётся с запасом. + AND toDate(created_at) BETWEEN {first_day:Date} - 1 AND {last_day:Date} +-- Тот же ноль, что в приёме заказов: заводская двойка исполнила бы вставку +-- локально на каждом шарде, минуя ключ шардирования цели, и строки унаследовали +-- бы раскладку источника. Ключи ODS и DDS здесь совпадают, и раскладка вышла бы +-- верной — но по совпадению, а не по контракту (docs/architecture/storage.md, +-- «Раскладка по шардам», капкан #137). +SETTINGS distributed_foreground_insert = 1, + parallel_distributed_insert_select = 0; diff --git a/sql/dds/order_replace.sql b/sql/dds/order_replace.sql new file mode 100644 index 0000000..3af9d70 --- /dev/null +++ b/sql/dds/order_replace.sql @@ -0,0 +1,28 @@ +-- DDS: подмена одного дня заказов собранной партицией двойника. +-- +-- Это и есть идемпотентность слоя. Повтор прогона безопасен, а «дыхание» +-- статусов внутри окна выражается сменой всей партиции, а не правкой строк. +-- +-- Замена идёт по локальным таблицам, шард за шардом: ON CLUSTER рассылает её +-- на обе ноды, а по Distributed то же самое не работает — операции с +-- партициями отвечают кодом 48 (docs/architecture/storage.md, «Что +-- проверено»). Правильность пошардовой замены держится на том, что двойник +-- шардирован тем же ключом, что цель, — см. sql/ddl/45-dds-tables.sql. +-- +-- Отсюда граница обещания: атомарность здесь пошардовая. Внутри шарда подмена +-- мгновенна — читатель видит либо прежний день целиком, либо новый; но две +-- локальные замены независимы, и в зазоре между ними чтение через Distributed +-- соберёт ответ из нового дня одного шарда и прежнего дня другого. Зазор +-- короткий, а отказ одной из двух замен растягивает его до повторного прогона. +-- +-- Пустая партиция двойника — законный случай, а не сбой: день, в котором +-- заказов нет, снимает день из цели, и слой перестаёт показывать то, чего в +-- источнике больше нет. +-- +-- День подставляет Jinja, а не параметр ClickHouse: в ALTER … PARTITION +-- сервер параметров не подставляет — замерено, отвечает синтаксической +-- ошибкой. Общее правило репозитория «параметры передаются отдельно, а не +-- строкой» здесь не нарушено по существу: значение пришло из ответа самого +-- ClickHouse на запрос объёма, а не снаружи стенда. +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 new file mode 100644 index 0000000..23fffa9 --- /dev/null +++ b/sql/dds/order_scope.sql @@ -0,0 +1,65 @@ +-- DDS: объём пересборки заказов — с какого дня по какой собирает этот запуск. +-- +-- Параметров у дага нет: объём он выводит сам из состояния слоёв +-- (docs/architecture/etl.md, «Объём прогона»). Ответ — две даты, границы +-- непрерывного отрезка; список дней между ними разворачивает даг. Пустой +-- источник даёт пустой ответ, и даг на нём останавливается с внятной ошибкой +-- вместо зелёного прогона, собравшего ничего. +-- +-- Файл кончается без точки с запятой: даг читает его целиком и шлёт одним +-- запросом, дописывая формат ответа, — точка с запятой сделала бы из этого +-- мультизапрос, который ClickHouse не принимает. +-- +-- Объём складывается из двух частей. Первая — дышащее окно: K последних дней +-- источника, состояние которых слепок ещё может изменить. Вторая — хвост от +-- первого дня, которого в слое нет: первая сборка застаёт слой пустым и +-- собирает весь стартовый мир, а после разгона мира пачкой слой догоняет +-- уехавший вперёд источник. +-- +-- **Хвост считается от первого недостающего дня, а не от последнего +-- собранного.** Разница не косметическая: замены партиций идут отдельными +-- задачами и падают порознь, поэтому упавший день оставляет дыру, пока +-- соседние дни собираются дальше. Считай мы прогресс по максимуму собранного +-- дня, дыра позади окна не заросла бы никогда, а прогоны продолжали бы +-- зеленеть над щербатым слоем. Цена честного ответа — проход по дню заказа в +-- ODS; она того стоит. +-- +-- Край окна берётся из делового поля, а не из ярлыков доставки: время +-- создания заказа одинаково у всех его версий, а `snapshot_date` и `_load_id` +-- выжившей строки достаются жребием (docs/architecture/etl.md, «Объём +-- прогона», и docs/architecture/orders/ingestion.md, «Роль ODS»). +-- +-- Известная неточность края: если в самый свежий день источника не родилось +-- ни одного заказа, окно съезжает на день назад — лишний замёрзший день +-- пересоберётся впустую, а пустой день не тронется. Данные от этого не +-- страдают, меняется только ярлык загрузки лишнего дня. Точный ответ здесь +-- дала бы дата слепка, но она и есть тот самый жребий. +WITH + -- FINAL не нужен: время рождения заказа версии не различают. + (SELECT max(toDate(toTimeZone(created_at, 'Europe/Samara'))) + FROM ods.order_dist) AS last_source_day, + (SELECT min(toDate(toTimeZone(created_at, 'Europe/Samara'))) + FROM ods.order_dist) AS first_source_day, + -- GLOBAL обязателен: обе таблицы распределённые, и обычный NOT IN сверял + -- бы дни источника с локальной частью слоя на каждом шарде — день, чьи + -- заказы легли на соседний шард, показался бы недостающим. + -- minOrNull отличает «дыр нет» от даты по умолчанию. + ( + SELECT minOrNull(day) + FROM ( + SELECT DISTINCT toDate(toTimeZone(created_at, 'Europe/Samara')) AS day + FROM ods.order_dist + ) + WHERE day GLOBAL NOT IN (SELECT order_date FROM dds.order_dist) + ) AS first_missing_day, + last_source_day - ({window_days:UInt16} - 1) AS window_start +SELECT + -- Дыр нет — начинаем с окна; есть — с первой из них, если она старше окна. + -- greatest держит левый край не раньше первого дня, известного источнику: + -- у молодого мира окно шире самого мира. + greatest( + least(window_start, coalesce(first_missing_day, window_start)), + first_source_day + ) AS first_day, + last_source_day AS last_day +WHERE (SELECT count() FROM ods.order_dist) > 0 -- 2.54.0 From 535b03783a914093ad068005436ba397eb17a768 Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Wed, 26 Aug 2026 05:32:03 -0400 Subject: [PATCH 2/2] =?UTF-8?q?docs(dds):=20=D1=83=D0=B1=D1=80=D0=B0=D0=BD?= =?UTF-8?q?=D1=8B=20=D0=BF=D0=BE=D0=B2=D1=82=D0=BE=D1=80=D1=8B=20=D0=BE?= =?UTF-8?q?=D0=B1=D1=8A=D1=8F=D1=81=D0=BD=D0=B5=D0=BD=D0=B8=D0=B9=20=D0=B2?= =?UTF-8?q?=20=D0=BA=D0=BE=D0=BC=D0=BC=D0=B5=D0=BD=D1=82=D0=B0=D1=80=D0=B8?= =?UTF-8?q?=D1=8F=D1=85=20=D0=B7=D0=B0=D0=BA=D0=B0=D0=B7=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Зачем: - Полный разбор урока про идемпотентность стоял дважды, предупреждение про окно K — трижды; внимание менти конечно, и каждый пересказ списывается с этого счёта. Что: - Докстринг dds_transform сокращён до одной фразы с указателями: полный разбор остался в шапке DDL, механика подмены — в order_replace.sql. - Комментарий про ORDER_WINDOW_DAYS в даге сведён к указателю на compose, где число объявлено и где записан риск расхождения. - Прорежен повтор одного риторического приёма «это не X» — оставлен там, где различие и есть суть. Проверка: - make lint, make config-test - Прогон dds_transform зелёный, слой сходится с ods.order_v: 1917 заказов, 8 дней. Co-Authored-By: Claude Opus 5 (1M context) --- compose.yaml | 4 ++-- dags/dds_transform.py | 23 +++++++---------------- sql/ddl/45-dds-tables.sql | 4 ++-- sql/dds/order_rebuild.sql | 4 ++-- sql/dds/order_scope.sql | 4 ++-- 5 files changed, 15 insertions(+), 24 deletions(-) diff --git a/compose.yaml b/compose.yaml index 70f2129..5a65ec0 100644 --- a/compose.yaml +++ b/compose.yaml @@ -54,8 +54,8 @@ x-world-starting-days: &world-starting-days "8" # генератора значило бы разрешить менять физику мира снаружи, а опись мира # держится на её неизменности. Стенду копия нужна затем, что по этому же числу # трансформация DDS решает, какие дни пересобирать. Разойдутся — пересборка -# промахнётся по краю окна, и заметит это не проверка, а менти со вчерашней -# выручкой. +# промахнётся по краю окна. Сверять копии некому, и это осознанно: слепые пятна +# конвейера собраны в docs/architecture/etl.md и адресованы контуру DQ. x-order-window-days: &order-window-days "7" x-airflow-common: &airflow-common diff --git a/dags/dds_transform.py b/dags/dds_transform.py index 39b854a..bae503f 100644 --- a/dags/dds_transform.py +++ b/dags/dds_transform.py @@ -8,17 +8,10 @@ дневного конвейера, с кнопки менти и после падения (docs/architecture/etl.md, «Объём прогона»). -Идемпотентность здесь устроена не так, как в приёме заказов, и в этом весь -урок сущности. Приём гасит повтор колонкой версии: ReplacingMergeTree -оставляет позднюю строку. Трансформация версий не знает — она собирает день -заново в таблицу-двойник и атомарно подменяет им партицию цели. Поэтому -повтор прогона не меняет делового содержимого слоя, а строки пересобранных -дней получают новый _load_id: видно, каким запуском собран каждый день. - -Замена партиции идёт по локальным таблицам с ON CLUSTER: по Distributed -операции с партициями отвечают кодом 48 (docs/architecture/storage.md, «Что -проверено»). Форма двойника и цена ошибки в ключе шардирования разобраны в -sql/ddl/45-dds-tables.sql. +Повтор прогона гасит не колонка версии, а замена партиции целиком: день +собирается в таблицу-двойник и подменяет собой партицию цели. Почему слой +устроен так — sql/ddl/45-dds-tables.sql; механика самой подмены — +sql/dds/order_replace.sql. """ from __future__ import annotations @@ -38,11 +31,9 @@ START_DATE = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC) SQL_ROOT = Path("/opt/airflow/sql") CLICKHOUSE_CONNECTION = "clickhouse_default" -# Окно изменяемости заказа, модельные дни. Число мира, а не настройка стенда: -# владелец — generator/world.py (ORDER_WINDOW_DAYS), который этим окном -# собирает слепок. Здесь оно читается, чтобы хранилище знало, какие дни ещё -# могут измениться. Разойдутся два объявления — пересборка промахнётся по краю -# окна, поэтому оба названы в compose.yaml рядом и с этой оговоркой. +# Окно изменяемости заказа, модельные дни: по нему трансформация решает, какие +# дни ещё могут измениться. Число принадлежит миру, владелец — генератор; +# почему у стенда лежит копия и чем она рискует — compose.yaml. ORDER_WINDOW_DAYS = int(os.environ["ORDER_WINDOW_DAYS"]) diff --git a/sql/ddl/45-dds-tables.sql b/sql/ddl/45-dds-tables.sql index edc19ee..9e22762 100644 --- a/sql/ddl/45-dds-tables.sql +++ b/sql/ddl/45-dds-tables.sql @@ -86,8 +86,8 @@ ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/{table}', '{ PARTITION BY order_date ORDER BY order_id; --- Двойник шардируется тем же ключом, что цель, и это не украшение: REPLACE --- PARTITION идёт по локальным таблицам, шард за шардом. Ляг строка заказа в +-- Двойник шардируется тем же ключом, что цель. REPLACE PARTITION идёт по +-- локальным таблицам, шард за шардом. Ляг строка заказа в -- доноре на другой шард, чем её место в цели, — подмена перенесла бы её не -- туда, и заказ пропал бы с одного шарда и задвоился на другом. CREATE TABLE IF NOT EXISTS dds.order_stage_dist ON CLUSTER clickstream_cluster diff --git a/sql/dds/order_rebuild.sql b/sql/dds/order_rebuild.sql index f886371..a43a053 100644 --- a/sql/dds/order_rebuild.sql +++ b/sql/dds/order_rebuild.sql @@ -10,8 +10,8 @@ -- раз на весь отрезок. Кто раскладывает строки по шардам — вставка через -- Distributed, ключ cityHash64(order_id) — разобрано в DDL двойника. -- --- Чистит двойник DROP PARTITION ALL, а не TRUNCATE, и это не вкусовщина: --- TRUNCATE — отдельное право, роли etl оно не выдано, а снятие всех партиций +-- Чистит двойник DROP PARTITION ALL, а не TRUNCATE: последнее — отдельное +-- право, роли etl оно не выдано, а снятие всех партиций -- входит в ALTER, который у неё есть (ADR 0007). Замерено: TRUNCATE отвечает -- ролью etl кодом 497, DROP PARTITION ALL проходит. Расширять права ради -- операции, для которой уже есть разрешённая форма, значило бы платить diff --git a/sql/dds/order_scope.sql b/sql/dds/order_scope.sql index 23fffa9..48a9db1 100644 --- a/sql/dds/order_scope.sql +++ b/sql/dds/order_scope.sql @@ -21,8 +21,8 @@ -- задачами и падают порознь, поэтому упавший день оставляет дыру, пока -- соседние дни собираются дальше. Считай мы прогресс по максимуму собранного -- дня, дыра позади окна не заросла бы никогда, а прогоны продолжали бы --- зеленеть над щербатым слоем. Цена честного ответа — проход по дню заказа в --- ODS; она того стоит. +-- зеленеть над щербатым слоем. Цена честного ответа — проход по дню заказа +-- в ODS. -- -- Край окна берётся из делового поля, а не из ярлыков доставки: время -- создания заказа одинаково у всех его версий, а `snapshot_date` и `_load_id` -- 2.54.0