Merge pull request 'feat(ods): версией заказа стала дата слепка, окно DDS — из данных' (#167) from 165-ods-order-version-snapshot-date into main

Reviewed-on: #167
This commit was merged in pull request #167.
This commit is contained in:
ddmitry committed 2026-08-26 18:50:37 +03:00
commit cdd44c469f
13 files changed
+152 -159

No files matched your search

-11
View File
@@ -48,16 +48,6 @@ x-kafka-target: &kafka-target
x-world-starting-days: &world-starting-days "8"
# Окно изменяемости заказа: слепок бэкенда несёт заказы последних K дней, за
# окном заказ замерзает. Число принадлежит модельному миру, и владелец у него
# один — generator/world.py (ORDER_WINDOW_DAYS): сделать окно настройкой
# генератора значило бы разрешить менять физику мира снаружи, а опись мира
# держится на её неизменности. Стенду копия нужна затем, что по этому же числу
# трансформация DDS решает, какие дни пересобирать. Разойдутся — пересборка
# промахнётся по краю окна. Сверять копии некому, и это осознанно: слепые пятна
# конвейера собраны в docs/architecture/etl.md и адресованы контуру DQ.
x-order-window-days: &order-window-days "7"
x-airflow-common: &airflow-common
image: clickstream-airflow:local
build:
@@ -87,7 +77,6 @@ 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 при разборе загружает
+1 -7
View File
@@ -19,7 +19,6 @@ from __future__ import annotations
import datetime
import logging
import os
from pathlib import Path
from airflow.exceptions import AirflowException
@@ -36,11 +35,6 @@ START_DATE = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC)
SQL_ROOT = Path("/opt/airflow/sql")
CLICKHOUSE_CONNECTION = "clickhouse_default"
# Окно изменяемости заказа, модельные дни: по нему трансформация решает, какие
# дни ещё могут измениться. Число принадлежит миру, владелец — генератор;
# почему у стенда лежит копия и чем она рискует — compose.yaml.
ORDER_WINDOW_DAYS = int(os.environ["ORDER_WINDOW_DAYS"])
# Позиция на оси модельного времени: номер первого несыгранного дня. Её ставит
# пульт мира, и только она отличает прожитый день от живого — зачем это
# сессиям, разобрано в sql/dds/session_scope.sql.
@@ -125,7 +119,7 @@ def dds_transform():
"""Спросить у хранилища, какие дни собирает этот запуск."""
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})
row = hook.get_first(query)
# Пустой ответ означает пустой источник: собирать нечего, и
# зеленеть на этом нельзя — прогон встаёт с внятной причиной.
if row is None:
+11 -12
View File
@@ -65,16 +65,15 @@ ODS: это единственный устойчивый ключ сущнос
одинаковы у всех строк дня и отвечают на вопрос «каким запуском и когда
собран этот день».
Ярлыки ODS — `snapshot_date`, `_load_id` приёма, `_load_ts` прибытия — в
модель не переезжают. Версию заказа они не задают, а физически выжившие их
значения — жребий: у строк с равным `updated_at` деловые поля совпадают
побайтово, и какая переживёт схлопывание — произвол (замер —
[приём заказов](../orders/ingestion.md), «Что проверено»). Перенесённый
ярлык выглядел бы фактом, оставаясь жребием; лживая координата хуже
отсутствующей. Отсюда правило слоя: **служебные колонки — координаты
собственной загрузки слоя; чужие координаты спрашиваются у их слоя по ключу
сущности**. Этим закрыто наследство #85: читатель у чужих ярлыков выше ODS
не нашёлся и при названных читателях модели.
Ярлыки ODS (`snapshot_date`, `_load_id` приёма, `_load_ts` прибытия) в модель
не переезжают. Они описывают наблюдение и доставку строки в ODS. Модель DDS
хранит текущее деловое состояние заказа. В ODS `snapshot_date` задаёт версию и
называет последний слепок, в котором источник показал заказ
([ADR 0014](../../adr/0014-order-version-snapshot-date.md)). Дату
наблюдения и координаты доставки читатель спрашивает у ODS по
`order_id`. Правило слоя:
**служебные колонки — координаты собственной загрузки слоя; чужие координаты
спрашиваются у их слоя по ключу сущности**.
Происшествие качества данных расследуется цепочкой, где каждый шаг
спрашивает свой слой на его языке: по `_load_id` строки DDS видно, какой
@@ -109,8 +108,8 @@ ODS: это единственный устойчивый ключ сущнос
позиций и пересчёт времён при каждом чтении собирали бы модель заново на
лету. Публичное `dds.order_v` поверх готовой физической модели этому доводу
не противоречит: оно ничего заново не вычисляет.
- **`ReplacingMergeTree(updated_at)` ещё раз.** Повторил бы механику ODS
вторым разом: физически хранил бы устаревшие состояния, вернул бы `FINAL`
- **`ReplacingMergeTree` ещё раз.** Повторил бы механику ODS вторым разом:
физически хранил бы устаревшие состояния, вернул бы `FINAL`
и обёртку-представление — а урок версий менти уже отыграл.
- **Отдельная сущность `dds.order_item`.** Вторая таблица, второй вопрос
шардирования и `JOIN` там, где хватает `ARRAY JOIN`; нормализация из
+28 -35
View File
@@ -5,10 +5,9 @@
движки, что именно пересобирается — живут в доках слоя
([карта модели DDS](dds/README.md)); здесь — только конвейер.
Спека обещала `etl_pipeline` этапом 5 — он понадобился раньше. Конвейер
рождается этапом 4, и временного в нём ничего нет: дневная сборка вперёд
останется и после. Этапу 5 достаётся его законный предмет — партиционная
переобработка, умение переиграть произвольный прошлый день по команде.
Дневная сборка вперёд появляется на этапе 4 и остаётся основным режимом
конвейера. На этапе 5 к ней добавляется партиционная переобработка: запуск
произвольного прошлого дня по команде.
## Форма
@@ -41,40 +40,37 @@
Параметров у конвейера нет: объём он выводит сам из `world_position` и
состояния слоёв. Сессии догоняют хвост — от первого несобранного дня по
последний прожитый; заказы пересобирают дышащее окно K и тот же хвост; карта
последний прожитый; заказы пересобирают дышащее окно и тот же хвост; карта
идентичностей пересчитывается целиком; правила витрин — у
[слоя DM](dm.md).
Поэтому один и тот же запуск одинаково верен из триггера, с кнопки менти и
после падения, а разгону на семь дней хватает одного прогона.
Окно заказов отсчитывается от **последнего дня, о котором рассказал сам
источник**, — от позднейшего дня заказа в ODS. Рядом напрашиваются ещё два
способа, и оба отвергнуты по одной причине: они отвечают не на тот вопрос.
Окно заказов задаёт последний слепок, наблюдаемый в ODS. У каждого заказа
`snapshot_date` равен последнему слепку, в котором источник его показал.
Трансформация выбирает строки с максимальной `snapshot_date` и берёт среди них
минимальный и максимальный `order_date`. Это границы дней, состояние которых
источник ещё может изменить
([ADR 0014](../adr/0014-order-version-snapshot-date.md)). Кроме живого окна,
трансформация закрывает пропуски: находит самый ранний день, представленный в
ODS и ещё не собранный в DDS, затем при необходимости расширяет до него левую
границу.
Позиция мира говорит, какой день прожит, а не какой доехал: прогон дня D везёт
слепок дня D − 1, да и позиция ставится раньше трансформации, так что после
неудачного приёма она обгоняет хранилище. Слой обязан считать объём по тому,
что у него есть, а не по тому, что мир о себе заявляет.
Позиция мира заказам не отвечает: она говорит, какой день прожит, а не какой
доехал. Прогон дня D везёт слепок дня D − 1, да и ставится позиция раньше
трансформации, так что после неудачного приёма обгоняет хранилище. Слой обязан
считать объём по тому, что у него есть, а не по тому, что мир о себе заявляет.
У сессий всё наоборот: им отвечает только позиция, потому что вопрос у них
другой. Им нужен не день, о котором рассказал источник, а день, который мир
**прожил**: события живой день от прожитого не отличают — и посреди дня D, и
сразу после его конца позднейший день событий равен D. Соберись сессии посреди дня, визит, начатый в
23:50, остался бы обрезанным навсегда: замёрзший день слой не пересобирает.
Отсюда правило — последний прожитый день равен первому дню событий плюс позиция
минус один ([`sql/dds/session_scope.sql`](../../sql/dds/session_scope.sql)).
Для сессий границу задаёт `world_position`: она отличает полностью прожитый
день от ещё идущего. Сами события этой границы не показывают: и посреди дня D,
и сразу после его конца их позднейший день равен D. Если собрать сессии посреди
дня, визит, начатый в 23:50, останется обрезанным: завершённый день слой
повторно не собирает. Поэтому последний прожитый день равен первому дню
событий плюс позиция минус один
([`sql/dds/session_scope.sql`](../../sql/dds/session_scope.sql)).
Оговорка про «доехал» остаётся в силе и здесь: за хвост потока отвечает не
позиция, а «Готовность источника».
Ярлыки доставки — `snapshot_date`, `_load_id`, `_load_ts` — не отвечают тоже, и
это свойство ODS, а не недосмотр: слой схлопывает доставки, сохраняя деловые
версии, поэтому у заказа, который не менялся, ярлык выжившей строки достаётся
жребием ([приём заказов](orders/ingestion.md), «Роль ODS»). Замер на стартовом
мире: последний слепок нёс все 1662 заказа, а его ярлык сохранили 1188 —
отбор «дни, у которых `snapshot_date` равен последнему» потерял бы четверть
заказов и молча пропустил бы дальний край окна, где заказы давно замёрзли.
К времени создания это не относится: у всех версий заказа оно одно.
## Готовность источника
Первым шагом `dds_transform` обязан дождаться, пока хвост топика `hits`
@@ -102,14 +98,11 @@
## Чего конвейер о себе не знает
Трансформация собирает слой по правилу, а не сверяет его с источником. Отсюда
четыре вещи, которых она не заметит, и у всех четырёх один адрес — контур DQ.
три вещи, которых она не заметит, и у всех трёх один адрес — контур DQ.
- **Окно K разъехалось с источником.** Число живёт двумя копиями: у генератора
в `world.py` и у стенда в `compose.yaml`. Разойдутся — слой начнёт
недобирать край окна или переписывать замёрзший день, и оба прогона будут
зелёными.
- **Пустой свежий день сдвигает край окна.** Если в позднейший день источника
не родилось ни одного заказа, окно съезжает на день назад
- **Часть последнего слепка не доехала.** Отказ вставки после коммита офсетов
уносит прочитанное с собой. Пропади так строки самого старого дня окна —
окно, выведенное из ярлыков, молча сузится, и день не пересоберётся
([`sql/dds/order_scope.sql`](../../sql/dds/order_scope.sql)).
- **Замена партиции прошла на одном шарде из двух.** День в слое есть — значит,
объём следующего прогона его не тронет, а слой останется собранным наполовину
+1 -1
View File
@@ -38,7 +38,7 @@
[мост к склейке](identity.md).
- **Приём — пакетный забор**: одно прямое чтение Kafka в STG, два
`INSERT SELECT` в типизированный ODS и таблицу ошибок; `ods.order`
принимает версии заказа на `ReplacingMergeTree(updated_at)` —
принимает версии заказа на `ReplacingMergeTree(snapshot_date)` —
[приём из Kafka в ODS](ingestion.md).
- **Дальше заказ живёт моделью DDS** — уже по хранилищную сторону границы,
в наборе слоя: [модель заказа](../dds/order.md).
+40 -27
View File
@@ -7,10 +7,10 @@
## Проблема
Заказы приезжают полным слепком окна изменяемости, но Kafka передаёт его
отдельными сообщениями и не сообщает потребителю, где слепок закончился. Прямое
чтение Kafka Engine возвращает одну порцию. Поэтому прежняя публикация через
`REPLACE PARTITION snapshot_date` приравнивала дату наблюдения к отсутствующей
транспортной границе и могла заменить день неполным набором строк.
отдельными сообщениями и не сообщает потребителю, где слепок закончился.
Прямое чтение Kafka Engine возвращает одну порцию без признака полноты.
Поэтому `snapshot_date` не может служить границей публикации: `REPLACE
PARTITION` могла бы заменить партицию дня неполным набором строк.
Одновременно типизированный ODS не должен принимать правдоподобные значения по
умолчанию из грязного JSON или останавливать весь пакет из-за одной строки.
@@ -19,8 +19,8 @@
- сохранить пакетный забор как контраст потоковому приёму событий;
- один раз принять байты в STG и независимо разложить строки на годные и брак;
- хранить в ODS типизированные версии заказов, не привязывая идемпотентность к
`snapshot_date`;
- хранить в ODS типизированные версии заказов, не публикуя слепок заменой
партиции;
- дать следующим слоям один корректный способ прочитать текущее состояние;
- оставить код проверки коротким и ограничить его контрактом провода.
@@ -81,6 +81,13 @@ JSON-объектом с точным набором ключей: `order_id`, `
малый контракт заказа нет причины. Граница строгости заканчивается на форме
провода: содержимое позиций и бизнес-инварианты намеренно остаются ниже.
Контракт источника задаёт частоту: **одной дате соответствует один слепок**. Это
свойство всего слепка, поэтому предикат одной строки его не проверяет. Вторая
выгрузка того же дня даст ничью по `snapshot_date`, и выбор строки потеряет
однозначность. Модельный источник выполняет условие; в настоящей интеграции
его фиксируют в договоре с источником
([ADR 0014](../../adr/0014-order-version-snapshot-date.md)).
Класс брака выбирается первым совпадением:
1. `not_an_object`;
@@ -94,31 +101,34 @@ JSON-объектом с точным набором ключей: `order_id`, `
## Роль ODS
`ods.order_rep` хранит физически принятые версии в
`ReplacingMergeTree(updated_at)`. Ключ сортировки — `order_id`, партиция — день
неизменного `created_at`. `ods.order_dist` шардирует по
`ReplacingMergeTree(snapshot_date)`. Ключ сортировки — `order_id`, партиция —
день неизменного `created_at`. `ods.order_dist` шардирует по
`cityHash64(order_id)`: только так все версии заказа попадают на один шард и
`FINAL` даёт корректный результат через распределённую таблицу.
Четыре координаты отвечают на разные вопросы:
- `updated_at` — какая бизнес-версия заказа новее;
- `snapshot_date` — в слепке какого модельного дня источник показал строку;
- `updated_at` — когда строку последний раз меняли в источнике;
- `_load_id` — какой запуск Airflow принял строку;
- `_load_ts` — когда строка приехала в хранилище.
Версию выбирает только `updated_at`. Заказ, не менявшийся несколько дней,
приезжает в нескольких слепках одинаковыми деловыми полями, и какая из этих
строк переживёт схлопывание — не обещано. Поэтому `snapshot_date` выжившей
строки — один из дней, когда источник показывал эту версию, не обязательно
последний; то же с `_load_id` и `_load_ts`. Вопрос «когда заказ наблюдался в
последний раз» задаётся сырью STG, пока то живо, — а не одной строке ODS.
Версию выбирает `snapshot_date`: выживает строка с максимальной датой слепка
([ADR 0014](../../adr/0014-order-version-snapshot-date.md)). Поэтому
`snapshot_date` показывает последнее наблюдение заказа. По строкам последней
даты слепка DDS определяет окно пересборки. `_load_id` и `_load_ts` принадлежат
той физической строке, которую сохранило слияние, и описывают её доставку.
Сама схема версий стоит на допущении об источнике: уточнение заказа поднимает
`updated_at`. Генератор моделирует источник с триггерной меткой и допущение
выполняет (замер — «Что проверено»); источник, умеющий уточнить строку тихо —
ручным `UPDATE` мимо приложения, миграцией, — ломает его, и тогда версии по
метке времени мало: нужна сверка самих полей, в терминах dbt snapshots —
стратегия `check` вместо `timestamp`.
Схема версий стоит на дисциплине источника: слепок с поздней датой не
показывает состояния старше. Модельный мир её держит — состояние читается на
границе суток, и слепок дня выходит чистой функцией зерна и дня; источник,
собирающий слепок из отставшей реплики, её ломает. Деловая метка `updated_at`
при этом монотонна по слепкам, поэтому выбор по дате слепка даёт тот же ответ,
что выбор по метке (замер — «Что проверено»). Тихая правка строки — ручным
`UPDATE` мимо приложения, миграцией — метку не двигает, но следующий слепок
перекроет её сам, пока заказ в окне изменяемости. За окном источник заказ
больше не показывает, и там помогает только сверка самих полей, в терминах
dbt snapshots — стратегия `check` вместо `timestamp`.
Обычное чтение `_dist` показывает физически сохранившиеся версии и нужно для
диагностики. Их число зависит от фоновых слияний: ODS не служит архивом истории.
@@ -201,12 +211,15 @@ JSON-объектом с точным набором ключей: `order_id`, `
`system.query_log` на первой ноде осталось ровно по одному завершённому запросу
пользователя `etl` для забора в STG, записи годных строк и записи брака.
Допущение и произвол ярлыков замерены 22 августа 2026 года при исполнении
#137: во всех 3034 парах строк с равным `updated_at` деловые поля совпали
побайтово, по слепкам метка ни разу не убывала, а `ods.order_v` сошёлся
построчно с эталоном `argMax` по `updated_at`, посчитанным прямо из сырья.
Произвол ярлыков виден въявь: у 86 заказов стартового мира выжила строка
серединного слепка — не первого и не последнего.
Дисциплина источника замерена 22 августа 2026 года при исполнении #137.
Во всех 3034 парах строк с равным `updated_at` деловые поля совпали
побайтово, а по слепкам метка ни разу не убывала.
Текущее состояние проверено на пересобранном мире 26 августа 2026 года при
исполнении #165. После двух слепков физическая таблица содержала 3356 версий
1917 заказов. У всех строк `ods.order_v` дата слепка совпала с последним
слепком заказа; деловые поля совпали с эталоном `argMax` по `updated_at` из
сырья. В обеих сверках расхождений нет.
Забор из Kafka в STG снят на живом стенде 18 августа 2026 года при исполнении
#93: одно прямое чтение приносит весь слепок дня, метаданные доставки доступны,
+12 -10
View File
@@ -183,8 +183,8 @@ kafka_offset)`: разбор полётов идёт от «какое сооб
заказов — пакетным шагом. Дальше метка переносится в ODS как есть и отвечает на
вопрос «когда строка приехала в хранилище», а не «когда её разобрали». В
`ods.event` она же служит колонкой версии `ReplacingMergeTree` и схлопывает
повтор доставки. У `ods.order` версию задаёт `updated_at` источника;
`_load_ts` только показывает, когда конкретная строка приехала.
повтор доставки. У `ods.order` версию задаёт `snapshot_date` — дата слепка
источника; `_load_ts` только показывает, когда конкретная строка приехала.
`created_at` и `updated_at` заказа к служебным колонкам хранилища не относятся.
Они приезжают в сообщении как аудит строки в БД источника и в ODS разбираются в
@@ -224,7 +224,7 @@ ODS и в таблицу ошибок. Нужен ли он выше ODS, реш
## Версии заказов
`ods.order_rep` хранит принятые версии в
`ReplacingMergeTree(updated_at)`: ключ сортировки — `order_id`, партиция —
`ReplacingMergeTree(snapshot_date)`: ключ сортировки — `order_id`, партиция —
`toDate(created_at)`. `ods.order_dist` шардирует по
`cityHash64(order_id)`. Все версии заказа лежат в одной партиции, чтобы их могли
схлопывать фоновые слияния, и на одном шарде, чтобы распределённый `FINAL`
@@ -234,11 +234,13 @@ ODS и в таблицу ошибок. Нужен ли он выше ODS, реш
версии, которые ещё не убрали фоновые слияния, и не является архивом истории.
`ods.order_v` служит поверхностью точного текущего состояния для следующих
слоёв. Представление сохраняет язык источника и не решает, какой станет модель
DDS. `snapshot_date` в нём остаётся датой наблюдения строки, а не ключом
публикации. Полное решение — в
[ADR 0010](../adr/0010-order-versions-in-ods.md) и
[спецификации приёма заказов](orders/ingestion.md). Какой моделью заказ
живёт дальше — [модель заказа в DDS](dds/order.md).
DDS. `snapshot_date` в нём одновременно дата наблюдения и колонка версии. Она
не доказывает полноту принятой порции, поэтому ODS не публикует партиции по
дате слепка. Модель версионного ODS записана в
[ADR 0010](../adr/0010-order-versions-in-ods.md), выбор колонки версии описан в
[ADR 0014](../adr/0014-order-version-snapshot-date.md), пакетный приём описан в
[спецификации приёма заказов](orders/ingestion.md). Контракт следующего слоя
задан в [модели заказа в DDS](dds/order.md).
## Часовые пояса
@@ -669,8 +671,8 @@ DDL-словарь с файловым источником внутри `user_f
порции из пяти строк (две версии одного заказа плюс по одной строке каждого
класса брака) годные ушли в `ods.order`, брак — в `ods.order_errors` с
ожидаемыми классами. Обе версии легли в одну партицию и на один шард; `FINAL`
через распределённую таблицу вернул одну строку — ту, у которой `updated_at`
позже, — тогда как физический счёт показывал две. Повтор задачи с тем же
через распределённую таблицу вернул одну строку — ту, что несла позднейшую
версию, — тогда как физический счёт показывал две. Повтор задачи с тем же
`_load_id` строку в `ods.order_v` и её `_load_ts` не изменил, а таблица ошибок
записала тот же брак второй раз, как и обещано спецификацией приёма.
@@ -126,6 +126,10 @@ parseDateTime64InJodaSyntaxOrNull(
использовать и для типизированной строки, и для маршрутизации ошибки; нулевая
дата не нужна.
> Актуальный стенд: версию `ods.order` задаёт `snapshot_date`
> ([ADR 0014](../adr/0014-order-version-snapshot-date.md)). Следующий
> абзац фиксирует исходный вывод исследования и не описывает текущий контракт.
`updated_at` допустим как колонка версии `ReplacingMergeTree`: ClickHouse
явно разрешает для `ver` тип `DateTime64` и оставляет строку с максимальной
версией
+11 -13
View File
@@ -220,10 +220,12 @@ Ecommerce (заполнены только у торговых событий):
[исследовании формата](../research/2026-08-16-order-snapshot-wire-format.md).
- `ods.order` принимает версии заказа в
`ReplacingMergeTree(updated_at)`: `ORDER BY order_id`, партиция по дню
неизменного `created_at`, шардирование по `cityHash64(order_id)`.
`snapshot_date` остаётся датой наблюдения источника, но не задаёт публикацию
или идемпотентность. Физическое чтение может видеть несколько версий;
`ReplacingMergeTree(snapshot_date)`: `ORDER BY order_id`, партиция по дню
неизменного `created_at`, шардирование по `cityHash64(order_id)`. Дата
слепка служит колонкой версии: выживает строка с максимальной
`snapshot_date` ([ADR 0014](../adr/0014-order-version-snapshot-date.md)).
Дата не доказывает полноту принятой порции, поэтому ODS не заменяет партицию
слепка целиком. Физическое чтение может видеть несколько версий;
`ods.order_v` возвращает точное текущее состояние. В DDS одна строка хранит
один заказ с позициями в типизированных массивах; физическая модель и
пакетное наполнение описаны в [доке заказа](../architecture/dds/order.md).
@@ -452,17 +454,13 @@ README.
матвью к нему не привязана, и сырьё забирает шаг, которым управляет Airflow —
одно прямое чтение вставляет порцию в `stg.orders_raw` с `_load_id` запуска.
Один следующий `task_id` двумя последовательными запросами пишет годные строки
и ошибки из того же среза. `ods.order` хранит версии по `updated_at`,
а не публикует партицию `snapshot_date`. Слой сырья остаётся точкой повтора и
разбора одной принятой порции.
Стенд получает от этого два режима приёма рядом, поток и слепок, и сравнение
из опорных точек раздела 12 переформулировано под них.
и ошибки из того же среза. `ods.order` выбирает текущую версию по максимальной
`snapshot_date` ([ADR 0014](../adr/0014-order-version-snapshot-date.md)). Слой
сырья остаётся точкой повтора и разбора одной принятой порции.
Состав служебных колонок задаёт дока хранилища. Спеке важны два следствия:
`ods.event` и `ods.order` получают метку загрузки `_load_ts`, но у
заказа бизнес-версию задаёт `updated_at`; `_load_id` проходит через STG и обе
цели ODS. Таблицы
`ods.event` и `ods.order` получают метку загрузки `_load_ts`, но версию
заказа задаёт дата слепка; `_load_id` проходит через STG и обе цели ODS. Таблицы
`stg.*_raw` хранят метаданные доставки Kafka вместе с именем читавшей ноды —
без них урок «какая нода читала топик» ненаблюдаем.
Модельного дня в STG нет: `EventDate` — свойство содержимого, а содержимое
+12 -10
View File
@@ -156,17 +156,19 @@ ENGINE = Distributed('clickstream_cluster', 'ods', 'event_errors_rep', cityHash6
-- Слепок привозит состояние заказов окна изменяемости, а не поток изменений,
-- и одна и та же сущность приезжает в нём много дней подряд. Поэтому строка
-- здесь — версия заказа, а не запись слепка: ключ сущности order_id, колонка
-- версии updated_at (время последнего изменения строки в источнике). Порядок
-- версий решает источник, а не хранилище.
-- версии snapshot_date — день, состояние которого снял источник. У полного
-- слепка версия и есть слепок: истинно то, что источник сказал последним
-- (ADR 0014).
--
-- Четыре координаты отвечают на разные вопросы, и путать их нельзя:
-- updated_at — какая бизнес-версия новее; snapshot_date — в слепке какого
-- модельного дня источник показал строку; _load_id — какой запуск Airflow её
-- принял; _load_ts — когда она приехала в хранилище. Ключом сущности не
-- становится ни одна из трёх последних: они про наблюдение и загрузку. Версию
-- выбирает только updated_at, поэтому у выжившей после схлопывания строки
-- snapshot_date — один из дней, когда источник показывал эту версию, не
-- обязательно последний (docs/architecture/orders/ingestion.md).
-- snapshot_date — в слепке какого модельного дня источник показал строку;
-- updated_at — когда строку последний раз меняли в источнике; _load_id —
-- какой запуск Airflow её принял; _load_ts — когда она приехала в хранилище.
-- Ключом сущности не становится ни одна: заказ один, а дней наблюдения и
-- версий у него много. Схлопывание оставляет строку позднейшего слепка,
-- поэтому snapshot_date выжившей — последний день, когда источник показывал
-- заказ, и по нему DDS читает объём пересборки
-- (docs/architecture/orders/ingestion.md).
--
-- PARTITION BY toDate(created_at) — по дню создания строки в источнике. Он
-- неизменен у всех версий заказа, поэтому версии лежат в одной партиции и
@@ -194,7 +196,7 @@ CREATE TABLE IF NOT EXISTS ods.order_rep ON CLUSTER clickstream_cluster
_load_id String,
_load_ts DateTime64(3, 'UTC')
)
ENGINE = ReplicatedReplacingMergeTree('/clickhouse/tables/{shard}/{database}/{table}', '{replica}', updated_at)
ENGINE = ReplicatedReplacingMergeTree('/clickhouse/tables/{shard}/{database}/{table}', '{replica}', snapshot_date)
PARTITION BY toDate(created_at)
ORDER BY order_id;
+4 -3
View File
@@ -11,9 +11,10 @@ FROM ods.event_dist FINAL;
-- ODS: каноническое чтение текущего состояния заказов.
--
-- То же, что у событий, но версию здесь задаёт источник: физическая таблица
-- хранит все приехавшие версии заказа, а FINAL оставляет ту, у которой
-- updated_at позже. Читать заказы голым SELECT по _dist значит зависеть от
-- того, сколько фоновых слияний успело пройти.
-- хранит все приехавшие версии заказа, а FINAL оставляет приехавшую
-- позднейшим слепком — колонка версии здесь snapshot_date. Читать заказы
-- голым SELECT по _dist значит зависеть от того, сколько фоновых слияний
-- успело пройти.
--
-- Поля остаются языком источника, обогащения нет: DDS читает это
-- представление и отдельно решает, какой будет его модель (ADR 0010).
+2 -2
View File
@@ -1,8 +1,8 @@
-- DDS: физические модели слоя и доноры их партиций.
--
-- Идемпотентность здесь устроена иначе, чем в приёме, и это главный урок обеих
-- сущностей файла. В ODS повтор гасит колонка версии:
-- ReplacingMergeTree(updated_at) оставляет позднюю строку. Здесь колонки
-- сущностей файла. В ODS повтор гасит колонка версии: ReplacingMergeTree по
-- snapshot_date оставляет строку позднейшего слепка. Здесь колонки
-- версии нет вовсе — повтор гасит замена партиции целиком: день собирается
-- заново и подменяется одним атомарным движением. Поэтому строка таблицы
-- ровно одна на своё зерно, а чтению не нужен FINAL.
+26 -28
View File
@@ -10,11 +10,11 @@
-- запросом, дописывая формат ответа, — точка с запятой сделала бы из этого
-- мультизапрос, который ClickHouse не принимает.
--
-- Объём складывается из двух частей. Первая — дышащее окно: K последних дней
-- источника, состояние которых слепок ещё может изменить. Вторая — хвост от
-- первого дня, которого в слое нет: первая сборка застаёт слой пустым и
-- собирает весь стартовый мир, а после разгона мира пачкой слой догоняет
-- уехавший вперёд источник.
-- Объём складывается из двух частей. Первая — дышащее окно: дни, которые нёс
-- последний слепок, то есть те, чьё состояние источник ещё может изменить.
-- Вторая — хвост от первого дня, которого в слое нет: первая сборка застаёт
-- слой пустым и собирает весь стартовый мир, а после разгона мира пачкой слой
-- догоняет уехавший вперёд источник.
--
-- **Хвост считается от первого недостающего дня, а не от последнего
-- собранного.** Разница не косметическая: замены партиций идут отдельными
@@ -24,22 +24,26 @@
-- зеленеть над щербатым слоем. Цена честного ответа — проход по дню заказа
-- в ODS.
--
-- Край окна берётся из делового поля, а не из ярлыков доставки: время
-- создания заказа одинаково у всех его версий, а `snapshot_date` и `_load_id`
-- выжившей строки достаются жребием (docs/architecture/etl.md, «Объём
-- прогона», и docs/architecture/orders/ingestion.md, «Роль ODS»).
-- Окно читается из ярлыка слепка, а не вычисляется вычитанием окна
-- изменяемости: у выжившей строки snapshot_date — последний слепок, который
-- нёс этот заказ (ADR 0014). Поэтому оба края окна знает сам источник, а
-- копии его константы на стороне стенда больше нет.
--
-- Известная неточность края: если в самый свежий день источника не родилось
-- ни одного заказа, окно съезжает на день назад — лишний замёрзший день
-- пересоберётся впустую, а пустой день не тронется. Данные от этого не
-- страдают, меняется только ярлык загрузки лишнего дня. Точный ответ здесь
-- дала бы дата слепка, но она и есть тот самый жребий.
-- FINAL не нужен ни здесь, ни ниже: время рождения заказа у всех его версий
-- одно, а строки последнего слепка несут максимум колонки версии — мерж
-- оставляет как раз их.
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,
(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,
-- GLOBAL обязателен: обе таблицы распределённые, и обычный NOT IN сверял
-- бы дни источника с локальной частью слоя на каждом шарде — день, чьи
-- заказы легли на соседний шард, показался бы недостающим.
@@ -51,15 +55,9 @@ WITH
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
) AS first_missing_day
SELECT
-- Дыр нет — начинаем с окна; есть — с первой из них, если она старше окна.
-- greatest держит левый край не раньше первого дня, известного источнику:
-- у молодого мира окно шире самого мира.
greatest(
least(window_start, coalesce(first_missing_day, window_start)),
first_source_day
) AS first_day,
last_source_day AS last_day
least(window_start, coalesce(first_missing_day, window_start)) AS first_day,
window_end AS last_day
WHERE (SELECT count() FROM ods.order_dist) > 0