chore(repo): влиты изменения задачи 155
- Зачем: - ветка ETL должна учитывать новую витрину `purchase_vs_orders` из `main`. - Что: - `origin/main` влит без переписывания истории опубликованного PR; - `dm_transform` теперь собирает третью витрину внутри общего конвейера. - Проверка: - `make config-test`, `make lint`; - `generator`: `make lint`, `make typecheck`, `make test` — 429 тестов; - `make rebuild-storage`, `make smoke`, `make check-clickhouse`, `make check-services`; - импорт DAG без ошибок, `purchase_vs_orders` заполнена.
This commit is contained in:
commit
cec11edaf3
15 files changed
+391
-38
No files matched your search
+96
-4
@@ -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,
|
||||
|
||||
@@ -45,6 +45,11 @@
|
||||
своё имя здесь создало бы третий синоним. `client_timezone` тоже остаётся —
|
||||
это атрибут посетителя, а не линза времени.
|
||||
|
||||
`purchase_revenue` сохраняет смысл поля источника. Тип денег нормализуется в
|
||||
ODS; точный тип и проверка описаны в разделе
|
||||
[«Актуальные события»](../storage.md#актуальные-события). DDS только
|
||||
переименовывает уже нормализованную колонку.
|
||||
|
||||
## Расшифровки
|
||||
|
||||
Одна: `device_category` — коды 1–4 в строки `desktop` / `mobile` /
|
||||
|
||||
+32
-2
@@ -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.
|
||||
|
||||
## Отклонённые варианты
|
||||
|
||||
|
||||
@@ -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 и переехали выше.
|
||||
@@ -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 товаров события |
|
||||
|
||||
@@ -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. Анонимность и склейка идентичностей
|
||||
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -27,10 +27,12 @@ PREAMBLE = """# Описание выгрузки: событие кликстр
|
||||
по элементу на товар. Отдельной сущности «визит» в выгрузке нет — визиты
|
||||
собирают на стороне хранилища, а `VisitID` дан как эталон для самопроверки.
|
||||
|
||||
Имена и типы колонок — стороны источника. Хранилище принимает их как есть и
|
||||
нормализует у себя: то же имя в нашем стиле ждёт в столбце «Нормализованное
|
||||
имя». Это имя источника, приведённое к snake_case, а не имя атрибута в
|
||||
модели данных: слой DDS складывает свою модель и называет атрибуты по ней.
|
||||
Имена и типы колонок описывают сторону источника. Обычно ODS принимает их без
|
||||
изменений. Исключение — `purchaseRevenue`: источник объявляет Float64, а ODS
|
||||
хранит деньги в Decimal. Точный локальный тип назван в документации
|
||||
хранилища. Имя в нашем стиле ждет в столбце «Нормализованное имя». Это имя
|
||||
источника, приведенное к snake_case, а не имя атрибута в модели данных: слой
|
||||
DDS складывает свою модель и называет атрибуты по ней.
|
||||
Столбец «Тип numpy» показывает, чем колонка представлена внутри генератора;
|
||||
у массивов это тип элемента. Номер — место колонки в выгрузке: порядок задан
|
||||
контрактом.
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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 и без партиций.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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;
|
||||
@@ -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;
|
||||
@@ -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
|
||||
Reference in new issue
Block a user