docs(ingestion): упрощены объяснения источников и приема данных

Зачем: помочь начинающему инженеру проследить событие и версию заказа
от генератора до хранилища, включая ошибки и повторные поступления.

Что: обновлены вводная генератора, справочники приема и модели заказа,
разделы хранилища и комментарии SQL и дага. Сохранены технические условия;
исполняемый код, схемы и поведение стенда не изменены.

Проверка: make lint, git diff --check, 142 локальные ссылки и справка CLI.
SQL без комментариев и Python AST без строк документации совпали
с исходными. Генерация данных и сброс состояния не выполнялись.

Refs #250
This commit is contained in:
ddadmin committed 2026-09-13 16:32:45 -04:00
1 parent 83434102d5
commit 75f28fe99f
13 files changed
+663 -974

No files matched your search

+86 -107
View File
@@ -1,134 +1,113 @@
# Модель заказа в DDS
Учебный результат: менти проектирует бизнес-сущность поверх готового ODS —
выбирает зерно, состав строки и способ наполнения — и видит идемпотентность
трансформации, устроенную иначе, чем в приёме: заменой партиции, а не
версиями.
DDS готовит заказ для анализа: выбирает текущее состояние из ODS,
разбирает позиции и переводит время в часовой пояс счетчика. Этот слой
показывает, как собрать модель данных для витрин и повторить загрузку
без удвоения заказов.
## Зерно
Одна строка — заказ в текущем состоянии, ключ — `order_id`. Модель отвечает
на вопрос «что сейчас известно о заказе»; историю версий не хранит:
читателя у истории нет ни в одной витрине, а ODS её и не обещает —
промежуточные версии съедают фоновые слияния. «Дыхание» выручки внутри окна
K живёт на текущем состоянии само: пересборка дня меняет статусы заказов, и
число дня меняется вместе с ними.
Одна строка — один заказ в последнем известном состоянии, ключ — `order_id`.
История изменений здесь не хранится. ODS тоже не служит ее архивом:
промежуточные версии удаляются при фоновых слияниях.
## Состав строки
Деловые поля источника переезжают из `ods.order_v` без переименований —
бэкенд уже говорит в snake_case: `order_id`, `user_id`, `status`, деньги в
`Decimal` (`items_total`, `discount`, `delivery`, `total`).
Поля заказа приходят из `ods.order_v` с именами источника: `order_id`,
`user_id`, `status`, `items_total`, `discount`, `delivery`, `total`.
Денежные колонки хранят рубли с копейками в `Decimal(18, 2)`.
Времена `created_at` и `updated_at` пересчитываются при загрузке в пояс
счётчика: DDS обслуживает человека с дашбордом ([дока
хранилища](../storage.md), «Часовые пояса»). Там же один раз считается
**день заказа** — колонка `order_date`, по образцу `EventDate` у события:
суточные срезы группируются по готовой дате и пояса не упоминают.
`created_at` и `updated_at` переводятся из UTC в `Europe/Samara`, пояс
счетчика сайта. Из `created_at` в этом поясе вычисляется `order_date` —
день оформления заказа. Витрины группируют по готовой дате;
правило времени общее для [хранилища](../storage.md#часовые-пояса).
Позиции разобраны из сырой строки `items` в три типизированных массива
одной длины — артикул, количество, цена (`Decimal`). Это родная для
ClickHouse денормализация: позиция не живёт без заказа и собственного ключа
не имеет, а разворачивает массивы читатель — `ARRAY JOIN` в витрине
выручки, с обогащением категорией через `dictGet`
([мастер-спека](../../specs/2026-07-30-stand-v2-realism.md), раздел 7). Разбор
JSON происходит один раз, на границе ODS → DDS, куда его и положила спека
(раздел 2); записывать его параллельными колонками-массивами или `Nested` —
выбор DDL, не модели.
Строка `items` из ODS разбирается в три массива одной длины:
| Колонка | Что хранит |
|---|---|
| `item_sku` | Артикулы товаров |
| `item_quantity` | Количество каждого товара, `UInt32` |
| `item_price` | Цены за единицу в рублях, `Decimal(18, 2)` |
Элементы с одинаковым номером описывают одну позицию заказа. Витрина
выручки разворачивает их через `ARRAY JOIN` и получает категорию товара
из справочника через `dictGet`. Разбор JSON выполняется один раз
при загрузке DDS.
ODS проверяет только то, что `items` — массив. Разбор в DDS не заменяет
проверку его содержимого: например, некорректное количество может стать
нулем. Проверку суммы позиций участник добавляет в
[первом задании ступени 5](../../labs/5-1-order-checks.md).
## Движок и наполнение
`dds.order_rep` — `ReplicatedMergeTree` без колонки версии: партиция —
`order_date` (день заказа неизменен), ключ сортировки — `order_id`.
`dds.order_dist` шардирует по `cityHash64(order_id)`, тем же выражением, что
ODS: это единственный устойчивый ключ сущности.
`dds.order_rep` — таблица на `ReplicatedMergeTree`, без колонки версии.
Партиция хранит один `order_date`, ключ сортировки — `order_id`.
Запись через `dds.order_dist` распределяет заказы по `cityHash64(order_id)`,
как в ODS. Для чтения служит `dds.order_v` с явным списком колонок
над `dds.order_dist`; `FINAL` в этом слое не нужен.
Идемпотентность держит замена партиции, а не версии. Сбор дня — батч после
прихода слепка: трансформация собирает партицию заново из `ods.order_v` и
атомарно подменяет. Поэтому каждая строка физической таблицы — ровно один
заказ, а чтению не нужен `FINAL`. Пока окно K живо, день «дышит» — статусы
заказов меняются, и партиция дня пересобирается очередным запуском; на выходе
из окна день замерзает.
Даг `dds_transform` пересобирает выбранные дни из `ods.order_v`:
Публичная поверхность — `dds.order_v`: обычное представление с явным списком
деловых и собственных служебных колонок над `dds.order_dist`, без `FINAL`.
Роли объектов и доступ следуют общей [конвенции](../storage.md), «Имена
объектов».
1. [`order_scope.sql`](../../../sql/dds/order_scope.sql) определяет
непрерывный отрезок `first_day`…`last_day`. Он включает дни заказов
из последнего слепка и начинается раньше, если в DDS отсутствует
день, который есть в ODS.
2. [`order_rebuild.sql`](../../../sql/dds/order_rebuild.sql) очищает
промежуточную таблицу `dds.order_stage_rep` и наполняет ее через
`dds.order_stage_dist`. Это донор: в нем готовится результат,
пока основная таблица остается доступной для чтения.
3. [`order_replace.sql`](../../../sql/dds/order_replace.sql) заменяет
каждую дневную партицию основной таблицы готовой партицией донора.
Механика запуска — где собирается партиция и как подменяется на кластере —
сюда не входит: это устройство [дневного конвейера](../etl.md).
Повтор при неизменном ODS дает те же деловые данные без дублей —
это идемпотентность загрузки. Пока заказы дня входят в семидневное окно
источника, их состояние может меняться, и день пересобирается.
За пределами окна новые состояния не приходят, но пропущенный день
можно восстановить из ODS. Границы пересборки берутся из данных;
число семь в SQL хранилища не повторяется.
Замена атомарна внутри одного шарда: читатель видит прежнюю или новую
партицию целиком. Замены на двух шардах независимы, поэтому между ними
запрос через Distributed может увидеть смешанный результат.
Донор остается до следующей сборки и доступен для диагностики.
Порядок задач и повторы — в [описании дневного конвейера](../etl.md).
## Служебные колонки
Строка получает `_load_id` и `_load_ts` запуска трансформации, собравшего
её партицию, — по общей конвенции хранилища. При замене партиции они
одинаковы у всех строк дня и отвечают на вопрос «каким запуском и когда
собран этот день».
`_load_id` и `_load_ts` показывают, какой запуск DDS собрал строку
и когда он выполнил вставку. Они одинаковы у строк одной сборки.
При следующей пересборке эти метки обновятся, даже если заказ не изменился.
Ярлыки ODS (`snapshot_date`, `_load_id` приёма, `_load_ts` прибытия) в модель
не переезжают. Они описывают наблюдение и доставку строки в ODS. Модель DDS
хранит текущее деловое состояние заказа. В ODS `snapshot_date` задаёт версию и
называет последний слепок, в котором источник показал заказ
([ADR 0014](../../adr/0014-order-version-snapshot-date.md)). Дату
наблюдения и координаты доставки читатель спрашивает у ODS по
`order_id`. Правило слоя:
**служебные колонки — координаты собственной загрузки слоя; чужие координаты
спрашиваются у их слоя по ключу сущности**.
`snapshot_date` и метки приема ODS в DDS не переносятся: они описывают
получение сообщения источника. Их можно найти в ODS по `order_id`.
В каждом слое служебные колонки относятся к его собственной загрузке.
Происшествие качества данных расследуется цепочкой, где каждый шаг спрашивает
свой слой на его языке: по `_load_id` строки DDS видно, какой запуск собрал
день; по `order_id` ODS показывает физические версии и брак — там свой
`_load_id` приёма связывает годные строки, ошибки и данные STG; дальше читают
STG, пока данные живы. Окно расследования задано сроками хранения: сырые данные
— трое суток, брак — месяц ([дока хранилища](../storage.md)).
При поиске ошибки начните с `_load_id` в DDS: он указывает на запуск
сборки. По `order_id` найдите версии заказа в ODS, а по их `_load_id` —
соответствующие сырые сообщения в STG и ошибки того же приема.
В таблице ошибок `order_id` может быть не разобран, поэтому нужное
сообщение ищут по `raw` и координатам Kafka.
Сырые сообщения хранятся около трех реальных суток, ошибки — месяц;
уточнения к срокам — в [справочнике хранилища](../storage.md#срок-жизни-сырых-данных).
## Читатели
- Витрина выручки `dm.revenue_daily_v`: заказы в `paid`, `ARRAY JOIN`
позиций, категория через `dictGet` (спека, раздел 7).
- Сверка `dm.purchase_vs_orders_v`: `order_id`, `order_date`, `items_total`,
`status`; соединение — GLOBAL JOIN с обеими сторонами-подзапросами
([исследование соединений](../../research/2026-08-22-distributed-join.md)).
- Сборка карты идентичностей: пара `order_id` ↔ `user_id`, источник —
`dds.order_v`.
## Не входит
- Механика запуска трансформации и замены партиций на кластере —
[дневной конвейер](../etl.md).
- Контур проверок качества (даг DQ, `dm.dq_summary`) — отдельный разговор
этапа 4.
- Инкрементальное наполнение вместо пересборки — техдолг #2.
## Отклонённые варианты
- **Одно динамическое представление вместо физической таблицы.** Рамка стенда —
показать модель данных, которую хранилище готовит потребителям; урок
«слой — контракт, а не копия» уже закреплён за `dds.event_v`, а разбор
позиций и пересчёт времён при каждом чтении собирали бы модель заново на
лету. Публичное `dds.order_v` поверх готовой физической модели этому доводу
не противоречит: оно ничего заново не вычисляет.
- **`ReplacingMergeTree` ещё раз.** Повторил бы механику ODS вторым разом:
физически хранил бы устаревшие состояния, вернул бы `FINAL`
и обёртку-представление — а урок версий менти уже отыграл.
- **Отдельная сущность `dds.order_item`.** Вторая таблица, второй вопрос
шардирования и `JOIN` там, где хватает `ARRAY JOIN`; нормализация из
реляционного мира, а не модель под колоночное хранилище. `JOIN` менти
увидит в сверке, работу с массивами — только здесь.
- **Зерно «версия заказа» (история).** Читателя нет; ODS историю не
обещает, а мир при нужде переигрывается генератором.
- **Перенос ярлыков ODS** — см. «Служебные колонки».
- **Без модели вовсе: витрины читают `ods.order_v` напрямую.** Разбор JSON
и пересчёт времён расползлись бы по витринам, а модель данных — то, ради
чего сущность и строится, — не была бы показана.
- `dm.revenue_daily_v` считает выручку по заказам со статусом `paid`,
разворачивает позиции и дополняет их категорией товара.
- `dm.purchase_vs_orders_v` сопоставляет заказ с покупкой на сайте
по `order_id`, используя `order_date`, `items_total` и `status`.
- `dds.identity_map` связывает идентификаторы посетителей с `user_id`
через покупки и заказы.
## Связанные решения
- [ADR 0010](../../adr/0010-order-versions-in-ods.md) — версии заказов в
ODS, устойчивая поверхность `ods.order_v`.
- [Приём заказов](../orders/ingestion.md) — граница строгого приёма и
четыре координаты строки ODS.
- [Проталкивание условий сквозь
представления](../../research/2026-08-22-view-predicate-pushdown.md) —
резолюция #128 той же карты: цена `FINAL` и цепочек представлений, фон, на
котором здесь выбрана физическая таблица.
- [ADR 0010](../../adr/0010-order-versions-in-ods.md) — версии заказов в ODS
и чтение через `ods.order_v`.
- [Прием заказов](../orders/ingestion.md) — проверки формата и выбор
актуальной версии.
- [Исследование соединений](../../research/2026-08-22-distributed-join.md) —
почему сверка распределенных источников использует `GLOBAL JOIN`.
- [Проталкивание условий сквозь представления](../../research/2026-08-22-view-predicate-pushdown.md) —
как фильтр запроса влияет на объем чтения ODS.
+135 -204
View File
@@ -1,241 +1,172 @@
# Приём заказов из Kafka в ODS
# Прием заказов из Kafka в ODS
Учебный результат: менти различает версию бизнес-сущности, наблюдение источника
и запуск загрузки, а затем читает физические версии через явную поверхность
текущего состояния.
## Проблема
Заказы приезжают полным слепком окна изменяемости, но Kafka передаёт его
отдельными сообщениями и не сообщает потребителю, где слепок закончился.
Прямое чтение Kafka Engine возвращает одну порцию без признака полноты.
Поэтому `snapshot_date` не может служить границей публикации: `REPLACE
PARTITION` могла бы заменить партицию дня неполным набором строк.
Одновременно типизированный ODS не должен принимать правдоподобные значения по
умолчанию из грязного JSON или останавливать весь пакет из-за одной строки.
## Цели
- сохранить пакетный забор как контраст потоковому приёму событий;
- один раз принять байты в STG и независимо разложить строки на годные и брак;
- хранить в ODS типизированные версии заказов, не публикуя слепок заменой
партиции;
- дать следующим слоям один корректный способ прочитать текущее состояние;
- оставить код проверки коротким и ограничить его контрактом провода.
## Не входит
- модель заказа в DDS: её зерно, связи, материализация и способ наполнения;
- проверка полей внутри `items`, переходов статуса и равенств денежных сумм;
- удаление заказа по отсутствию в следующем слепке;
- маркер конца слепка, опись ожидаемых строк и транзакция между целями ODS.
Здесь разобран путь одного заказа: сообщение Kafka сохраняется в STG,
проверяется и попадает либо в ODS, либо в таблицу ошибок. Следующие слои
читают из ODS последнюю принятую версию каждого заказа.
## Поток данных
После завершения генератора Airflow один раз читает байтовый Kafka-чтец и
записывает полученную порцию в `stg.orders_raw`. Даг зовётся `orders_ingest`, а
запускает его шаговый даг модельного времени, который положил слепок в топик и
ждёт конца прогона. Все строки получают `_load_id`, равный `run_id` Airflow.
`_load_ts` вычисляется при этой записи и дальше переносится без пересчёта.
Запрос забора лежит в
[`sql/stg/orders_raw_load.sql`](../../../sql/stg/orders_raw_load.sql):
`SQLExecuteQueryOperator` задаёт подключение и передаёт `run_id` как параметр
ClickHouse `load_id`, а преобразование остаётся в SQL своего слоя.
Источник отправляет в топик `orders` дневной слепок: по одному сообщению
на заказ. Слепок дня D содержит состояния заказов, оформленных в дни
D−6…D, с учетом [задержек первого появления](fate.md).
В начале модельного времени окно короче. Выгрузку за D отправляет
генератор при запуске дня D+1; подробности — в [описании слепка](snapshot.md).
Один следующий `task_id` отвечает за весь переход STG → ODS. Внутри него два
последовательных `INSERT SELECT` читают неизменный срез по `_load_id`: первый
пишет годные строки в `ods.order`, второй — брак в `ods.order_errors`.
Транзакции между запросами нет. При частичном сбое Airflow повторяет весь
`task_id`; одинаковые исходные строки и служебные метки не вычисляются заново.
Обе вставки лежат в одном самодостаточном файле
[`order_parse.sql`](../../../sql/ods/order_parse.sql). Airflow сначала
шаблонизирует файл, затем делит получившийся SQL на выражения и исполняет их по
порядку в одной задаче `parse_batch`.
После отправки шаговый даг модельного времени запускает
[`orders_ingest`](../../../dags/orders_ingest.py) и ждет его завершения.
У `orders_ingest` нет расписания; его два шага выполняются по порядку:
Условия запросов взаимодополняющие: один общий предикат определяет годную
строку, а ветвь брака использует его буквальное отрицание. Все функции предиката
возвращают результат без исключения, а сам предикат всегда заканчивается в
`true` или `false`, не в `NULL`. Постоянный классификатор между STG и ODS для
этого не нужен. Предикат объявлен локальным макросом Jinja внутри
`order_parse.sql` и разворачивается в оба выражения. Макрос убирает
дублирование, но не вводит отдельный фрагмент, который нельзя исполнить
самостоятельно.
1. `pull_batch` читает порцию из `stg.orders_raw_kafka` и записывает ее
через `stg.orders_raw_dist`. Содержимое сообщения остается строкой `raw`,
рядом сохраняются координаты Kafka, `_load_id` и `_load_ts`.
2. `parse_batch` читает сохраненные строки с этим `_load_id`. Первый
`INSERT SELECT` пишет годные записи в `ods.order_dist`, второй —
остальные в `ods.order_errors_dist`.
`_load_id` равен `run_id` Airflow и связывает строки одного запуска.
`_load_ts` — реальное время записи в STG; при разборе оно переносится
без пересчета. Запросы лежат в
[`orders_raw_load.sql`](../../../sql/stg/orders_raw_load.sql) и
[`order_parse.sql`](../../../sql/ods/order_parse.sql).
`SQLExecuteQueryOperator` передает `run_id` параметром ClickHouse `load_id`.
Две вставки разбора используют одно условие годности, записанное макросом
Jinja внутри `order_parse.sql`. Перед выполнением макрос подставляется
в оба запроса; ветвь брака берет отрицание этого условия. Оно возвращает
только истину или ложь и не вызывает исключений на некорректном `raw`.
Так каждая строка сохраненной порции получает ровно один результат.
Транзакции между вставками нет: первая может завершиться, а вторая упасть.
Повтор `parse_batch` выполняет обе вставки над тем же срезом STG.
Текущее состояние заказа останется прежним, но таблица ошибок может
получить повторные строки с теми же `_load_id` и координатами Kafka.
Обе вставки ждут доставки строк на шарды и используют ключ целевой таблицы;
настройки объяснены в [справочнике хранилища](../storage.md#раскладка-по-шардам).
## Граница строгого приёма
Единица решения — одна строка `stg.orders_raw`. Корень должен быть
JSON-объектом с точным набором ключей: `order_id`, `user_id`, `status`,
`created_at`, `updated_at`, `items_total`, `discount`, `delivery`, `total`,
`items`, `snapshot_date`.
Проверяется одна строка `stg.orders_raw_dist`. Корень JSON должен быть
объектом с точным набором из одиннадцати ключей:
`order_id`, `user_id`, `status`, `created_at`, `updated_at`,
`items_total`, `discount`, `delivery`, `total`, `items`, `snapshot_date`.
Скалярные поля проверяются по типу JSON. Деньги дополнительно обязаны быть
строками неотрицательной суммы с ровно двумя знаками после точки — минус в эту
форму не входит; времена — строками RFC 3339 в UTC с обязательными
миллисекундами; дата слепка — строкой `YYYY-MM-DD`. `items` проверяется только
как JSON-массив. Каноническая форма и основания выбора зафиксированы в
[исследовании формата](../../research/2026-08-16-order-snapshot-wire-format.md).
| Поля | Условия приема |
|---|---|
| `order_id`, `status` | Строки JSON |
| `user_id` | Целое число JSON, которое разбирается в `UInt64` |
| `created_at`, `updated_at` | Строки UTC с миллисекундами, например `2026-06-01T12:34:56.000Z`; проверяются форма и разбор времени |
| `items_total`, `discount`, `delivery`, `total` | Неотрицательные суммы в рублях, записанные строками с двумя знаками после точки; проверяется разбор в `Decimal(18, 2)` |
| `snapshot_date` | Строка даты `YYYY-MM-DD`, которая проходит разбор |
| `items` | Массив JSON; сохраняется целиком в строковой колонке |
Проверять все верхнеуровневые поля здесь уместно: их одиннадцать, и десять
скалярных значений непосредственно образуют типизированную строку заказа. У
события из 47 полей проверяются только пять опорных; переносить то сокращение на
малый контракт заказа нет причины. Граница строгости заканчивается на форме
провода: содержимое позиций и бизнес-инварианты намеренно остаются ниже.
ODS проверяет формат верхнего уровня. Здесь нет проверки допустимых
статусов, их переходов, полей внутри `items` или равенства денежных сумм.
Позиции разбираются в [DDS](../dds/order.md); проверку их суммы участник
добавляет в [задании ступени 5](../../labs/5-1-order-checks.md).
Основания выбора формата — в [исследовании](../../research/2026-08-16-order-snapshot-wire-format.md).
Контракт источника задаёт частоту: **одной дате соответствует один слепок**. Это
свойство всего слепка, поэтому предикат одной строки его не проверяет. Вторая
выгрузка того же дня даст ничью по `snapshot_date`, и выбор строки потеряет
однозначность. Модельный источник выполняет условие; в настоящей интеграции
его фиксируют в договоре с источником
([ADR 0014](../../adr/0014-order-version-snapshot-date.md)).
Класс ошибки записывается в `error_class` по первому нарушенному условию:
Класс брака выбирается первым совпадением:
1. `not_an_object` — корень JSON не объект;
2. `keyset_mismatch` — набор ключей отличается от ожидаемого;
3. `field_invalid` — тип, форма или разбор значения не прошли проверку.
1. `not_an_object`;
2. `keyset_mismatch`;
3. `field_invalid`.
Имя отдельного поля в класс не включается. В таблице ошибок остаются сырой
текст, метаданные доставки и `_load_id`, поэтому единичный случай можно разобрать
без постоянной детализации предиката.
Имя поля в класс не входит. Для разбора конкретной ошибки в
`ods.order_errors_dist` есть исходный `raw`, координаты Kafka и метки загрузки.
## Роль ODS
`ods.order_rep` хранит физически принятые версии в
`ReplacingMergeTree(snapshot_date)`. Ключ сортировки — `order_id`, партиция —
день неизменного `created_at`. `ods.order_dist` шардирует по
`cityHash64(order_id)`: только так все версии заказа попадают на один шард и
`FINAL` даёт корректный результат через распределённую таблицу.
`ods.order_rep` использует `ReplicatedReplacingMergeTree(snapshot_date)`.
Строки с одним ключом сортировки `order_id` считаются версиями одного заказа;
остается строка с максимальной `snapshot_date`. Партиция —
`toDate(created_at)` в UTC, ключ шардирования — `cityHash64(order_id)`.
Неизменный `created_at` и общий ключ направляют все версии заказа
в одну партицию и на один шард.
Четыре координаты отвечают на разные вопросы:
| Колонка | На какой вопрос отвечает |
|---|---|
| `snapshot_date` | На конец какого модельного дня источник показал это состояние? |
| `updated_at` | Когда запись заказа последний раз меняли в магазине? |
| `_load_id` | Какой запуск Airflow принял сообщение? |
| `_load_ts` | Когда сообщение записали в STG по реальным часам стенда? |
- `snapshot_date` — в слепке какого модельного дня источник показал строку;
- `updated_at` — когда строку последний раз меняли в источнике;
- `_load_id` — какой запуск Airflow принял строку;
- `_load_ts` — когда строка приехала в хранилище.
Например, заказ не менялся с 1 июня, но входит в слепки за 1 и 2 июня.
`updated_at` у строк одинаков, а `snapshot_date` различается. ODS оставит
строку за 2 июня вместе с метками ее доставки. Новая дата наблюдения
нужна DDS для определения окна пересборки, даже если состояние не изменилось.
Версию выбирает `snapshot_date`: выживает строка с максимальной датой слепка
([ADR 0014](../../adr/0014-order-version-snapshot-date.md)). Поэтому
`snapshot_date` показывает последнее наблюдение заказа. По строкам последней
даты слепка DDS определяет окно пересборки. `_load_id` и `_load_ts` принадлежат
той физической строке, которую сохранило слияние, и описывают её доставку.
Источник выпускает **один слепок на дату**. Повторная отправка дает то же
содержимое, а более поздний слепок не возвращает устаревшее состояние.
Эти условия не проверяются на одной строке при приеме. Если источник
пришлет два разных состояния с одной `snapshot_date`, эта колонка
не позволит определить, какое из них актуально.
Обоснование версии — [ADR 0014](../../adr/0014-order-version-snapshot-date.md).
Схема версий стоит на дисциплине источника: слепок с поздней датой не
показывает состояния старше. Модельный мир её держит — состояние читается на
границе суток, и слепок дня выходит чистой функцией зерна и дня; источник,
собирающий слепок из отставшей реплики, её ломает. Деловая метка `updated_at`
при этом монотонна по слепкам, поэтому выбор по дате слепка даёт тот же ответ,
что выбор по метке (замер — «Что проверено»). Тихая правка строки — ручным
`UPDATE` мимо приложения, миграцией — метку не двигает, но следующий слепок
перекроет её сам, пока заказ в окне изменяемости. За окном источник заказ
больше не показывает, и там помогает только сверка самих полей, в терминах
dbt snapshots — стратегия `check` вместо `timestamp`.
`ods.order_v` читает `ods.order_dist FINAL` и возвращает одну текущую строку
на `order_id`, даже если фоновые слияния еще не убрали старые версии.
Без `FINAL` физическая таблица может показать несколько строк заказа.
Историю ODS не гарантирует: старые версии постепенно удаляются при слияниях.
Отсутствие заказа в новом слепке не удаляет его из ODS.
Обычное чтение `_dist` показывает физически сохранившиеся версии и нужно для
диагностики. Их число зависит от фоновых слияний: ODS не служит архивом истории.
`ods.order_v` сохраняет те же источник-ориентированные поля без обогащения
данными модели, но возвращает одну актуальную версию на `order_id`. Сначала оно
может быть простым представлением над `_dist FINAL`; способ выбора можно
заменить, не меняя потребителей.
Это представление остаётся ответственностью ODS: оно скрывает механику чтения
версий, но не строит бизнес-модель. DDS читает `ods.order_v` и отдельно решает,
какие сущности, связи и производные признаки ему нужны. Судьба служебных
колонок выше ODS решена вместе с моделью DDS: чужие ярлыки туда не переезжают
([модель заказа](../dds/order.md)).
DDS читает `ods.order_v`, переводит времена в пояс счетчика и разбирает
позиции. Для своей загрузки DDS записывает собственные служебные метки —
[модель заказа](../dds/order.md#служебные-колонки).
## Одно чтение Kafka
Стандартный слепок содержит около полутора тысяч строк, тогда как предел одной
порции на стенде — десятки тысяч сообщений. Поэтому один запуск Airflow делает
один прямой `SELECT`, без цикла до пустоты и без фиксации конечных офсетов.
Один запуск делает один прямой `SELECT`, без чтения в цикле до пустоты.
Обычный слепок содержит около полутора тысяч строк, а предел порции
на стенде — десятки тысяч сообщений. Это допущение о размере данных:
Kafka не сообщает запросу, что прочитан весь слепок.
Это допущение о размере стенда, а не доказательство полноты слепка. Если Kafka
вернёт короткую порцию, непрочитанный хвост останется в топике и приедет в один
из следующих запусков. После отказа от замены партиции это задержка, а не потеря
или публикация неполного дня.
Короткая порция оставляет непрочитанные сообщения для следующих запусков.
Поэтому ODS добавляет версии заказов по мере поступления. Замена партиции
по `snapshot_date` могла бы опубликовать неполный набор заказов
([ADR 0010](../../adr/0010-order-versions-in-ods.md)).
Отказ — случай другой, и «хвост дождётся» на него не распространяется. Офсеты
порции коммитятся в момент чтения, поэтому упавшая вставка уносит прочитанное с
собой: повторное чтение вернёт ноль, а часть строк может уже лежать на шарде.
Позиция мира при этом не двигается, и тот же день уедет заново — в STG он
окажется частичным дублем, а старый хвост, ушедший той же порцией, не вернётся
([ADR 0008](../../adr/0008-order-ingestion.md), «Следствия»).
## Отклонённые варианты
- Партиционная идемпотентность — `REPLACE PARTITION snapshot_date` или
`ReplacingMergeTree` по `(snapshot_date, order_id)`: у потребителя нет
признака полноты партиции, а дата наблюдения становится частью ключа
сущности.
- Обычный `MergeTree` в ODS с дедупликацией только в DDS: навсегда сохраняет
технические повторы там, где семантика версии уже известна.
- Маркер, опись, чтение до пустоты или конечные офсеты: добавляют протокол ради
объёма, который с большим запасом помещается в одну порцию.
- Два `task_id` или материализованный классификатор: дробят один короткий
переход слоя, не добавляя транзакционности.
- Представление текущего состояния в DDS и готовая схема `dds.order` в этой
задаче: перекладывают механику ODS на следующий слой и преждевременно задают
модель данных.
## Риски и проверка
- На малой управляемой порции дать по одной строке каждого класса брака и две
годные версии одного `order_id`. Две цели должны сохранить все непустые
сообщения, а `ods.order_v` — вернуть новую версию независимо от фонового
слияния.
- Повторить переход с тем же `_load_id`: строка в `ods.order_v` и её `_load_ts`
не должны измениться; версии одного заказа должны остаться на одном шарде и
в одной партиции. Таблица ошибок может снова записать тот же брак: совпавшие
`_load_id` и Kafka-координаты показывают повтор задачи.
Сбой после чтения опаснее короткой порции. Чтец подтверждает позиции
сообщений в Kafka — офсеты — при прямом чтении. Если последующая вставка
в STG упадет, эти сообщения уже не будут прочитаны той же группой повторно;
часть из них может успеть записаться на шарды. Позиция модельного времени
при неуспехе не продвигается. Повтор шагового дага снова отправит слепок
того же дня, но не восстановит старые сообщения других слепков,
если они попали в неудачную порцию.
Граница восстановления описана в [ADR 0008](../../adr/0008-order-ingestion.md#следствия).
## Что проверено
Переход STG → ODS снят на живом стенде 18 августа 2026 года при исполнении #94.
Опыты этого раздела прогнаны и подтвердили обещанное: разбиение строк STG на
годные строки и брак полное и непересекающееся; две версии одного заказа легли в
одну партицию и на один шард, а `ods.order_v` вернуло позднюю независимо от
фонового слияния; повтор задачи с тем же `_load_id` строку в `ods.order_v` и её
`_load_ts` не изменил, а таблица ошибок записала тот же брак второй раз; строки
опытов убраны. Числа, механика опытов и поведение ClickHouse, на которое всё это
опирается, — [документ хранилища](../storage.md), «Что проверено».
Это результаты прежних опытов на стенде; при редактуре они заново
не выполнялись.
После исправления
[#102](https://git.dementev.space/ddmitry/clickstream-data-platform/issues/102)
честный прогон дня снят на живом стенде 20 августа 2026 года: все 1694 строки
слепка дня 7 приняты, `ods.order_errors` осталась пустой. У заказа
`20260608-0007`, который прежде приходил с отрицательным итогом, теперь
`5490.00 − 823.50 + 299.00 = 4965.50`.
- 18 августа 2026 года (#93, #94) прямое чтение приняло слепок,
а разбор разделил строки на годные и ошибки без потерь и пересечений.
Две версии заказа попали на один шард в одну партицию; `ods.order_v`
вернул позднюю. Повтор разбора с тем же `_load_id` сохранил текущую строку
и ее `_load_ts`, но повторил ошибки. Подробности —
[справочник хранилища](../storage.md#что-проверено) и
[ADR 0008](../../adr/0008-order-ingestion.md#что-проверено).
- 20 августа после исправления
[#102](https://git.dementev.space/ddmitry/clickstream-data-platform/issues/102)
приняты все 1694 строки слепка дня 7 без ошибок. Итог заказа
`20260608-0007`: `5490.00 − 823.50 + 299.00 = 4965.50`.
- 21 августа (#124) пересборка через `SQLExecuteQueryOperator` завершила
`pull_batch` и `parse_batch`; в `system.query_log` осталось по одному
завершенному запросу `etl` на забор, годные строки и ошибки.
- 22 августа (#137) во всех 3034 парах с одинаковым `updated_at`
деловые поля совпали побайтово; по слепкам время изменения не убывало.
- 26 августа (#165) после двух слепков физическая таблица содержала
3356 версий 1917 заказов. В `ods.order_v` у всех заказов была дата
последнего слепка, а деловые поля совпали с `argMax` по `updated_at`
из сырых данных.
При исполнении #124 21 августа 2026 года адресная пересборка мира прошла через
`SQLExecuteQueryOperator`: `pull_batch` и `parse_batch` завершились успешно. В
`system.query_log` на первой ноде осталось ровно по одному завершённому запросу
пользователя `etl` для забора в STG, записи годных строк и записи брака.
Дисциплина источника замерена 22 августа 2026 года при исполнении #137.
Во всех 3034 парах строк с равным `updated_at` деловые поля совпали
побайтово, а по слепкам метка ни разу не убывала.
Текущее состояние проверено на пересобранном мире 26 августа 2026 года при
исполнении #165. После двух слепков физическая таблица содержала 3356 версий
1917 заказов. У всех строк `ods.order_v` дата слепка совпала с последним слепком
заказа; деловые поля совпали с эталоном `argMax` по `updated_at` из сырых
данных. В обеих сверках расхождений нет.
Забор из Kafka в STG снят на живом стенде 18 августа 2026 года при исполнении
#93: одно прямое чтение приносит весь слепок дня, метаданные доставки доступны,
офсеты коммитятся, а сбой забора не двигает позицию мира. Числа — [ADR
0008](../../adr/0008-order-ingestion.md), раздел «Что проверено».
MCP Context7 в сессии проектирования был недоступен. На локальном ClickHouse
`26.3.17.56` проверено, что прямой `SELECT` Kafka Engine завершается после
одной порции, а `FINAL` через `Distributed` исполняется на таблицах шардов.
Поэтому версии одного `order_id` направляются на один шард. Фоновое схлопывание
`ReplacingMergeTree` и необходимость точного чтения сверены с
[официальной документацией](https://clickhouse.com/docs/reference/engines/table-engines/mergetree-family/replacingmergetree),
поведение чтения — с исходниками той же версии
На ClickHouse `26.3.17.56` отдельно проверены одна порция при прямом
`SELECT` Kafka и выполнение `FINAL` на шардах через Distributed.
При проектировании Context7 был недоступен; чтение сверяли с исходниками
той же версии:
[`StorageKafka.cpp`](https://github.com/ClickHouse/ClickHouse/blob/v26.3.17.56-lts/src/Storages/Kafka/StorageKafka.cpp) и
[`KafkaSource.cpp`](https://github.com/ClickHouse/ClickHouse/blob/v26.3.17.56-lts/src/Storages/Kafka/KafkaSource.cpp).
Повторная сверка правил версий с документацией — в
[справочнике хранилища](../storage.md#актуальные-события).
## Связанные решения
+205 -228
View File
@@ -122,112 +122,110 @@ DDS: представление читает события ODS.
## Служебные колонки
Собственные колонки не повторяют имён виртуальных. Виртуальные даёт движок:
`_topic`, `_partition`, `_offset`, `_timestamp` у Kafka, `_shard_num` у
`Distributed` и прочие. Если положить на диск колонку с таким же именем, в
матвью перестанет читаться, что дано движком, а что положено нами, — а это
ровно то различие, ради которого метаданные доставки и хранятся. Поэтому они
ложатся под именами `kafka_topic`, `kafka_partition`, `kafka_offset`,
`kafka_timestamp`, рядом — `consumer_host`, имя читавшей ноды: виртуальные
колонки его не несут, а после записи в `Distributed` он уже невосстановим.
STG сохраняет исходное сообщение в строке `raw` и добавляет сведения
о доставке. JSON на этом шаге не разбирается: некорректное содержимое
тоже попадает в таблицу.
Типы у них такие: `kafka_topic` и `consumer_host` — `LowCardinality(String)`,
значений мало и они повторяются; `kafka_partition` и `kafka_offset` — `UInt64`.
С `kafka_timestamp` сложнее, и форма его решена на стенде. Меток времени движок
даёт две: `_timestamp` — `Nullable(DateTime)`, то есть секунды, и
`_timestamp_ms` — `Nullable(DateTime64(3))`, миллисекунды. Колонка объявлена
`Nullable(DateTime64(3, 'UTC'))` и заполняется из `_timestamp_ms`: у брокера
метка миллисекундная, соседняя `_load_ts` тоже миллисекундная, а сырой слой
хранит приехавшее, и округлять ему нечего. Обнуляемость нужна отдельно от
разрядности: брокер метку заполняет не всегда, а необнуляемый тип значил бы либо
падение приёма на первом сообщении, либо тихие нули за 1970 год.
| Колонка STG | Что хранит |
|---|---|
| `kafka_topic` | Имя топика из виртуальной колонки Kafka `_topic` |
| `kafka_partition` | Номер партиции из `_partition` |
| `kafka_offset` | Позицию сообщения внутри партиции из `_offset` |
| `kafka_timestamp` | Метку времени Kafka из `_timestamp_ms`, если она есть |
| `consumer_host` | Имя ноды, прочитавшей Kafka |
| `_load_ts` | Реальное время записи сообщения в STG |
| `_load_id` | Только у заказов: `run_id` запуска Airflow, который принял порцию |
Заполняются все они выражением в `SELECT` матвью приёма, а не `DEFAULT` в
таблице. Для `consumer_host` это обязательно: `DEFAULT hostName()` вычисляется
на шарде-получателе, то есть назвал бы не ту ноду, которая читала топик, —
колонка молча отвечала бы на другой вопрос.
Топик, партиция и смещение вместе позволяют найти сообщение Kafka.
`kafka_topic` и `consumer_host` имеют тип `LowCardinality(String)`,
`kafka_partition` и `kafka_offset` — `UInt64`.
`kafka_timestamp` имеет тип `Nullable(DateTime64(3, 'UTC'))`: сохраняет
миллисекунды и допускает отсутствие метки.
Само сообщение лежит в колонке `raw` тем, чем пришло: чтец читает байты и ничего
не проверяет, поэтому там оказываются и целые сообщения, и мусор. События ниже
разбирают матвью ODS ([ADR 0005](../adr/0005-event-ingestion.md)), заказы —
пакетный шаг ([ADR 0008](../adr/0008-order-ingestion.md), [ADR
0010](../adr/0010-order-versions-in-ods.md)).
Виртуальные колонки предоставляет движок, а колонки STG хранят значения
на диске. Их имена различаются, чтобы не скрыть виртуальную колонку
собственной. Значения вычисляются в запросе приема. В частности,
`hostName()` должен выполняться на ноде, читающей Kafka:
`DEFAULT hostName()` только в локальной таблице назвал бы шард-получатель.
Движок таблицы STG — обычный `ReplicatedMergeTree`, `ORDER BY (kafka_partition,
kafka_offset)`: разбор полётов идёт от «какое сообщение», другого ключа у
таблицы STG и нет. Замену версий сюда ставить нельзя — она отменила бы свойство
слоя, ради которого он заведён: повтор доставки в STG обязан быть виден.
Таблицы STG используют `ReplicatedMergeTree` с
`ORDER BY (kafka_partition, kafka_offset)`. Повторная доставка остается
отдельной строкой; STG нужен в том числе для разбора таких повторов.
Метка времени загрузки зовётся `_load_ts`, тип `DateTime64(3, 'UTC')`. Ставится
она один раз при записи в STG: для событий — матвью приёма, для дневного слепка
заказов — пакетным шагом. Дальше метка переносится в ODS как есть и отвечает на
вопрос «когда строка приехала в хранилище», а не «когда её разобрали». В
`ods.event` она же служит колонкой версии `ReplacingMergeTree` и схлопывает
повтор доставки. У `ods.order` версию задаёт `snapshot_date` — дата слепка
источника; `_load_ts` только показывает, когда конкретная строка приехала.
`_load_ts` имеет тип `DateTime64(3, 'UTC')`. При переходе в ODS метка
копируется из STG без пересчета. У событий она также выбирает последнюю
доставку; у заказов версию задает `snapshot_date`. `_load_id` заказа
переносится в годную строку ODS и в таблицу ошибок.
У непрерывного потока событий нет запуска Airflow, поэтому `_load_id`
в его STG и ODS отсутствует.
`created_at` и `updated_at` заказа к служебным колонкам хранилища не относятся.
Они приезжают в сообщении как аудит строки в БД источника и в ODS разбираются в
`DateTime64(3, 'UTC')`; ClickHouse их не создаёт и добавляет рядом собственную
`_load_ts`. Совпадение слов «техническое время» не делает эти часы одной осью.
Время события `UTCEventTime` и дата `EventDate` приходят из счетчика сайта.
Поля заказа `created_at` и `updated_at` приходят из базы магазина.
Они относятся к модельному времени; `_load_ts` — к реальным часам стенда.
Например, событие июня можно загрузить в сентябре: его `EventDate`
останется июньской, а `_load_ts` будет сентябрьской.
Пакетной переобработки у событий в ODS нет: слой наполняют матвью, а не задание
Airflow. Переделать разобранное руками можно вставкой из STG с фильтром по
`_load_ts` в пределах трёхсуточного окна; ничья по версии разрешается в пользу
вставленного позже. Заказы разбирает пакетный шаг из неизменного среза STG по
`_load_id`; дневные партиции ODS он не заменяет.
Имя согласовано с каноном служебных полей соседнего учебного стенда на
Greenplum, чтобы словарь был общим у двух хранилищ; ведущее подчёркивание у
технических колонок — распространённая запись,
её же используют Fivetran, Airbyte и Stitch. С правилом выше это не спорит:
запрещено совпадать с именами виртуальных колонок, а не носить подчёркивание.
Общего идентификатора пачки загрузки (`_load_id`) нет. У потока событий нет ни
пачки, ни `run_id`, и колонка была бы пустой формальностью. У заказов читатель
назван: `_load_id` равен `run_id` Airflow и переносится из STG в годную строку
ODS и в таблицу ошибок. Нужен ли он выше ODS, решается вместе с моделью DDS.
События разбираются материализованными представлениями. Автоматической
пакетной переобработки у них нет; сохраненный срез STG можно разобрать
отдельным запросом до истечения срока хранения. При повторной вставке
с тем же ключом и `_load_ts` `ReplacingMergeTree` оставляет более поздно
вставленную строку. Заказы разбирает задача Airflow по `_load_id`.
В DDS метки загрузки уже относятся к собственной сборке слоя —
[пример заказа](dds/order.md#служебные-колонки).
## Актуальные события
`ods.event_rep` хранит физически принятые версии события в
`ReplacingMergeTree(_load_ts)`, а `ods.event_dist` служит для записи и
диагностики доставки. Обычное чтение этой пары может показать несколько строк
одного `WatchID`, пока фоновые слияния их не схлопнули.
Повторная отправка события может дать несколько строк в `ods.event_rep`.
Движок `ReplicatedReplacingMergeTree(_load_ts)` считает версиями строки
с одинаковым полным ключом сортировки:
`(CounterID, EventDate, intHash32(ClientID), WatchID)`.
При слиянии остается строка с максимальным `_load_ts`, то есть последняя
доставка. Один `WatchID` обозначает событие, но физический ключ замены шире.
`ods.event_v` — обычное представление над `ods.event_dist FINAL`. Оно сохраняет
имена, типы и зерно события, но всегда возвращает актуальную версию. Актуальное
состояние читают через него, а не повторяют `FINAL` в каждом запросе.
`dds.event_v` отвечает за другое: даёт событию язык бизнес-модели, в том
числе snake_case-имена и расшифровку кодов.
`ods.event_v` читает `ods.event_dist FINAL`: выбирает актуальную строку
при выполнении запроса, не дожидаясь фоновых слияний. Поэтому события
считают через `_v`; число строк в `_dist` зависит от уже прошедших слияний.
`dds.event_v` использует этот результат, дает колонкам имена в `snake_case`
и расшифровывает коды.
Типы ODS обычно повторяют контракт выгрузки. Денежный `purchaseRevenue` —
исключение: источник объявляет `Array(Float64)`, а материализованное
представление ODS разбирает числа сырого JSON сразу в `Array(Decimal(18, 2))`.
DDS наследует этот тип без нового преобразования. Прямой разбор подтвержден
документацией ClickHouse через Context7 и запросом к ClickHouse 26.3; сравнение
с сырыми данными подтвердило: копейки всех покупок целы.
Выбор по полному `ORDER BY` и максимальной версии, включая чтение с `FINAL`,
сверен через Context7 с [документацией ClickHouse](https://clickhouse.com/docs/reference/engines/table-engines/mergetree-family/replacingmergetree).
В стенде закреплена версия 26.3.17.56; `_v` использует `FINAL`, чтобы
результат не зависел от фоновых слияний.
Типы ODS повторяют [описание выгрузки](../formats/clickstream-event.md)
с одним денежным исключением: `purchaseRevenue` объявлен источником
как `Array(Float64)`, а из JSON разбирается сразу в
`Array(Decimal(18, 2))`. Это сумма в рублях с копейками; DDS наследует тип.
`productPrice` содержит округленные целые рубли и не подходит для
восстановления точной суммы покупки. Прямой разбор `purchaseRevenue`
проверен на ClickHouse 26.3; результаты сохранены в разделе «Что проверено».
## Версии заказов
Слепки повторяют состояние заказов из семидневного окна источника.
`ods.order_rep` хранит принятые версии в
`ReplacingMergeTree(snapshot_date)`: ключ сортировки — `order_id`, партиция —
`toDate(created_at)`. `ods.order_dist` шардирует по
`cityHash64(order_id)`. Все версии заказа лежат в одной партиции, чтобы их могли
схлопывать фоновые слияния, и на одном шарде, чтобы распределённый `FINAL`
выбрал одного победителя.
`ReplicatedReplacingMergeTree(snapshot_date)`: ключ сортировки — `order_id`,
партиция — `toDate(created_at)` в UTC. Запись через `ods.order_dist`
распределяет их по `cityHash64(order_id)`. У всех версий заказа одинаковы
день создания и шард, поэтому они могут заменять друг друга.
Физическая пара нужна для загрузки и диагностики. Обычное чтение показывает
версии, которые ещё не убрали фоновые слияния, и не является архивом истории.
`ods.order_v` служит поверхностью точного текущего состояния для следующих
слоёв. Представление сохраняет язык источника и не решает, какой станет модель
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).
`ods.order_v` читает `_dist FINAL` и оставляет строку с максимальной датой
слепка. Дата `snapshot_date` показывает последнее принятое наблюдение,
а `updated_at` — время изменения заказа в магазине. У неизменившегося
заказа в следующем слепке дата наблюдения обновится, время изменения — нет.
Источник выпускает один слепок на дату, а поздний слепок показывает
не менее новое состояние. Отсутствие заказа в следующем слепке
не означает удаления. Физическая таблица не служит архивом:
старые версии исчезают при фоновых слияниях.
Прием добавляет версии отдельными строками. `snapshot_date` не доказывает,
что Kafka отдала весь слепок, поэтому партиция по этой дате не заменяется.
Подробности — [прием заказов](orders/ingestion.md),
[ADR 0010](../adr/0010-order-versions-in-ods.md) и
[ADR 0014](../adr/0014-order-version-snapshot-date.md).
Следующий слой описан в [модели заказа DDS](dds/order.md).
## Часовые пояса
@@ -289,169 +287,148 @@ UTC+4), и пересчёт идёт один раз при наполнении
## Приём: поток и его свойства
Раздел — про события. У заказов приём устроен иначе: слепок забирает по команде
даг `orders_ingest`, а не подписанная навсегда матвью — [ADR
0008](../adr/0008-order-ingestion.md) и [спецификация приёма
заказов](orders/ingestion.md).
События принимаются непрерывно. Заказы читает отдельным запросом
даг `orders_ingest` — [описание пакетного приема](orders/ingestion.md).
У событий путь такой:
Цепочка одна: чтец топика → матвью → STG → матвью разбора → событие и таблица
ошибок ODS.
```text
stg.hits_raw_kafka → stg.hits_raw_mv → stg.hits_raw_dist
├→ ods.event_mv → ods.event_dist
└→ ods.event_errors_mv → ods.event_errors_dist
```
Чтец стоит на обеих нодах и читает одной группой потребителей — имя группы
`clickstream_hits`, и оно одинаково на обеих нодах по построению: DDL идёт
`ON CLUSTER` и макросов в имени не содержит. Разные группы дали бы каждой ноде
полную копию топика, и это отдельная сцена для лабы, а не рабочий режим. Имя
кластера в `ON CLUSTER` и в движке `Distributed` — `clickstream_cluster`, оно
задано в `infra/clickhouse/config.d/cluster.xml`.
Таблицы Kafka на обеих нодах используют одну группу `clickstream_hits`.
Ноды делят чтение партиций топика. Разные группы прочитали бы каждая
свою полную копию потока. Имя группы одинаково в DDL на всем
`clickstream_cluster`; конфигурация кластера —
`infra/clickhouse/config.d/cluster.xml`.
**Источник матвью разбора — `stg.hits_raw_dist`, а не локальная таблица.**
У матвью две привязки: источник, на вставку в который она срабатывает, и цель,
куда пишет. Распределённая таблица — лицо слоя, локальная — его хранилище;
потребитель слоя цепляется к лицу. Практически это значит, что разбор идёт на
той же ноде, что читала Kafka, в момент вставки первой матвью — до раскладки по
шардам.
Материализованные представления разбора срабатывают на вставку
в `stg.hits_raw_dist`, до распределения строк по шардам.
Оба видят вставляемую порцию на той ноде, которая читает Kafka.
`stg.hits_raw_mv` создается после них: его подключение запускает
чтение топика. Если начать чтение раньше, сообщения успеют попасть
в STG без разбора в ODS.
**Вставка фоновая, и окно потери мы принимаем.** Вставка в распределённую
таблицу кладёт блок в локальный спул и сразу возвращает управление, а Kafka
коммитит офсеты по факту работы матвью — то есть по факту записи в спул. Топик
уже считает сообщение прочитанным, хотя на шарде его ещё нет: умри нода в этом
промежутке — сообщения не перечитаются.
Условие приема проверяет три вещи:
Закрывает окно настройка `distributed_foreground_insert = 1`, и на ETL-вставках
Airflow она стоит — там это обычный `SETTINGS` у запроса. На пути приёма её нет,
и по трём причинам. Вставку выполняет фоновый поток Kafka-движка, своего запроса
у него не бывает, так что настройка уровня запроса доехала бы только профилем
пользователя в конфигурации ноды. Синхронный режим связывает шарды: пока второй
недоступен, вставка падает, офсеты не коммитятся, и приём встаёт целиком — тогда
как при фоновом первая нода продолжает принимать и копит спул для соседа.
Платится при этом не одно ожидание на блок, а три распределённые вставки — STG,
событие, ошибки, — и все внутри потока-потребителя, что само по себе повод для
ребаланса по таймауту сессии. Против всего этого — окно в сотню миллисекунд на
ноутбуке, где мир пересобирается одной командой. Размен не в пользу настройки, а
компромисс полезнее показать, чем спрятать за галочкой.
- корень `raw` — объект JSON;
- набор верхнеуровневых ключей совпадает с 47 полями формата;
- пять опорных полей, `WatchID`, `VisitID`, `ClientID`, `EventDate`
и `UTCEventTime`, разбираются в ожидаемые типы.
Ещё одно место, где сырой слой хранит не всё приехавшее: запись с пустым
значением и запись-надгробие проходят молча, не оставляя строки. Свойство
измерено и принято осознанно — подробности в разделе «Что проверено».
Остальные поля извлекаются по описанию выгрузки без отдельных проверок.
Это граница приема: годная строка не означает, что проверен смысл
всех 47 значений. Подробности — в [ADR 0005](../adr/0005-event-ingestion.md).
**Гарантии нет ни в одну сторону — есть два узких окна.** Окно потери описано
выше: нода умерла между коммитом офсетов и сбросом спула. Окно дубля
противоположное: нода умерла после записи на шард, но до коммита офсетов, и при
перечитывании сообщение приедет второй раз. Сказать про такой приём «хотя бы
один раз» нельзя — это обещало бы, что потерь не бывает, а они возможны.
`ods.event_mv` принимает строки, для которых условие истинно,
`ods.event_errors_mv` — для которых оно ложно. Условие не должно
возвращать `NULL`: такую строку не выбрали бы ни оно, ни его отрицание.
Разбор использует функции без исключений на некорректном сообщении,
поскольку ошибка материализованного представления прерывает вставку
и останавливает чтение Kafka до устранения причины.
Дубль ниже по течению ведёт себя по-разному. В ODS его схлопнет
`ReplacingMergeTree`, а таблицы STG вовсе не устраняют дубли: перезаливка
модельного дня честно удваивает `count()` в STG, и живёт эта пара до истечения
срока хранения. Это свойство слоя, а не поломка, — но обещание идемпотентности
конвейера к сырому слою не относится.
### Доставка и повторы
Оговорка к последнему: у семейства `Replicated*` есть своя дедупликация — блок с
тем же хешем, вставленный повторно, отбрасывается (`insert_deduplicate`).
Удвоение строк STG проходит мимо неё только потому, что при повторном чтении
`_load_ts` новый и хеш блока другой. Свойство слоя держится на этом, а не на
отсутствии механизма.
Вставка в Distributed на этом пути фоновая: блок сначала записывается
в локальную очередь на диске, затем отправляется на шарды.
Kafka подтверждает обработку после выполнения материализованных
представлений, когда данные еще могут находиться в очереди.
Если после подтверждения потерять ноду вместе с очередью, сообщения
не перечитаются. Возможен и повтор: данные уже записались на шард,
но подтверждение Kafka не успело сохраниться.
Поэтому прием не гарантирует доставку каждого сообщения хотя бы один раз.
**Матвью разбора две, и их условия обязаны делить поток без зазора и без
нахлёста.** Одна забирает годные строки в `ods.event_dist`, вторая — брак в
`ods.event_errors_dist`. Строка, подошедшая обеим, задвоится; не подошедшая ни
одной — исчезнет молча. Держится это формой: второе условие пишется буквальным
отрицанием первого, а сам предикат NULL не возвращает ни в одной своей части, —
иначе трёхзначная логика даст строку, которую не возьмёт ни `условие`, ни
`NOT условие`. Обнуляемый разбор в предикате поэтому есть, но заканчивается
`IS NOT NULL`, а сравнения дают 0 или 1. Функции предиката не должны и бросать
исключений: упавшая матвью
роняет вставку и останавливает потребление до починки
([ADR 0005](../adr/0005-event-ingestion.md)).
В ETL-запросах Airflow `distributed_foreground_insert = 1` дожидается
записи на шарды. У фонового приема событий этой настройки нет:
недоступность одного шарда тогда остановила бы весь прием. В принятом
режиме доступная нода продолжает принимать и накапливает очередь
для недоступного шарда. Ограничение сохранено для учебного стенда;
подробности и наблюдения — в разделе «Что проверено».
Постоянной сверки **слоёв между собой** при этом нет и не должно быть. У сырого
слоя срок жизни трое суток, а ODS хранит всё, поэтому равенство «строки STG =
события + ошибки» разъедется само: строки STG истекут раньше, а перезаливка
модельного дня задвоит их, тогда как в ODS тот же повтор схлопнется. Правило,
красное в норме, учит не смотреть на оповещения ([ADR
0002](../adr/0002-monitoring-scope.md)). Равенство проверено разовым опытом при
исполнении #43, на управляемой пачке: отправили N сообщений — получили N строк
STG и N в сумме событий и ошибок. Постоянной целью такой опыт не становится, и
почему — в [карте проверок](testing.md), раздел «Интеграционная проверка
постоянной целью не становится». Счёт актуальных событий идёт через
`ods.event_v`, которое скрывает `FINAL`: голый `count()` по `ReplacingMergeTree`
зависит от того, сколько мержей успело пройти, и спека это прямо запрещает
(раздел 6).
В STG повтор сохраняется, в ODS его устраняет выбор актуальной версии.
Хотя `Replicated*` умеет отбрасывать повторно вставленные одинаковые блоки,
при новом чтении Kafka меняется `_load_ts`, а вместе с ним и хеш блока.
Поэтому повторная отправка модельного дня увеличивает число строк STG.
**Постоянная сверка одна, и она другого рода** — не слой против слоя, а ODS
против [паспорта мира](../../data/world-inventory.json), лежащего в git (`make
check-clickhouse`, пришла с #42). Разъехаться сама она не может, и в этом вся
разница: паспорт описывает восемь дней стартового мира, счёт обрамлён их датами,
повтор заливки схлопывается в `ods.event_v`, а срок хранения ей не помеха — ODS
хранит всё. Обе стороны равенства зафиксированы: одна кодом генератора, другая
файлом в git. У межслойной сверки такой опоры нет ни с одной стороны.
Пустое значение сообщения и значение `null` не дают строки даже в STG:
чтец продвигает позицию Kafka, но не передает строку на разбор.
Генератор таких сообщений не создает. Поведение измерено на стенде
и описано в разделе «Что проверено».
### Что можно сверять
Для короткого неизменного среза доставки можно проверить баланс:
число строк STG равно сумме годных строк и ошибок до удаления повторов.
Постоянно сравнивать полные таблицы так нельзя: STG удаляет старые данные,
ODS оставляет события и объединяет повторные доставки.
Счетчики по слоям тогда расходятся при штатной работе.
`make check-clickhouse` сравнивает актуальные события в `ods.event_v`
с [паспортом мира](../../data/world-inventory.json) только за восемь
начальных дней. Эта проверка учитывает повторную отправку через `FINAL`
и не зависит от срока жизни STG. Границы остальных проверок —
в [карте проверок](testing.md).
## Срок жизни сырых данных
Сырые данные в STG живут трое суток реального времени и уходят сами. Трое — это
окно отладки: столько сырые данные хранятся на ноутбуке, чтобы менти успел
разобрать полёты, после чего перестаёт занимать место. Нарезка — по дню
загрузки, срок — по той же колонке `_load_ts`, снятие — целыми кусками
(`ttl_only_drop_parts`). В DDL значение проставлено явно, чтобы поведение не
зависело от умолчания версии.
TTL — правило автоматического удаления старых данных. В обеих таблицах
STG срок задан как три суток от `_load_ts`, по реальному времени загрузки.
Партиции тоже нарезаны по дню `_load_ts`. Это оставляет короткий период
для поиска ошибок и повторного разбора, после которого данные освобождают
место на диске.
Нарезать сырые данные по модельному дню события было бы соблазнительно — он
единица переобработки и он же ключ партиции в ODS, — но чистку это ломает. Кусок
снимается целиком, только когда в нём истекли все строки, а в партиции
модельного дня лежит приехавшее в разное реальное время: мерж склеит куски
разного возраста, самая свежая строка удержит весь кусок, и данные переживут
срок неограниченно.
Настройка `ttl_only_drop_parts = 1` удаляет только целые части данных —
файлы, которые ClickHouse создает при вставках и объединяет при слияниях.
Часть удаляется, когда срок истек у всех ее строк. Внутри партиции
дня загрузки возраст строк отличается не более чем на сутки.
Поэтому к трем суткам может добавиться до суток ожидания самой свежей
строки части, а затем ожидание фоновой обработки TTL. Точного времени
удаления правило не обещает.
В партиции дня загрузки склейка идёт точно так же, и строки в ней тоже разного
возраста — но не более чем на сутки, потому что партицию закрывает календарный
день. Отсюда и оценка: сырые данные живут трое суток плюс хвост до суток, а не
ровно трое. Оговорка про сроки: TTL исполняется на мержах, а не по будильнику,
так что «уходит само» здесь обещано, а «уходит вовремя» — нет.
Модельная дата для TTL не подходит: события июня могут загружаться
осенью, и срок от `EventDate` истек бы сразу. Партиция по модельному дню
тоже смешала бы доставки разного возраста: очередная отправка дня могла бы
продлевать хранение старых строк после слияния частей.
Модельного дня среди колонок STG нет вовсе. Он свойство содержимого, а
содержимое разбирает ODS — там `EventDate` и живёт, ключом партиции. Сырые
данные режутся по своим координатам: и разбор полётов, и переобработка фильтруют
по `_load_ts`, попадая в ключ партиции, а не идя сплошным проходом. Фильтр по
модельному дню вдобавок пропускал бы битые строки — у них дата не извлекается.
Две оси времени тут не совпадают намеренно. Ось модельного времени начинается в
D0 и к реальному календарю не привязана; пакетный режим проигрывает две недели
модельного мира за минуты реальных. Поэтому у переобработки и у гигиены диска
разные часы, и обслуживают их разные средства. Декларативный TTL по модельной
дате был бы просто сломан: он отсчитывает срок от реального «сейчас» и удалял бы
эталонные дни прямо на входе.
В бою сырой слой иногда собирают на движке `Null` — тогда он не хранится вовсе.
Такой вариант отвергнут: на стенде сырые данные нужны для отладки, поэтому окно,
а не ноль.
Для разбора сохраненных сообщений выбирайте небольшой срез по `_load_ts`.
Так запрос ограничивает партиции и сохраняет некорректные сообщения,
из которых модельную дату извлечь нельзя. `EventDate` появляется
отдельной колонкой только в ODS. В [задании ступени 6](../labs/6-1-event-parsing.md)
такой срез разбирается рядом с действующим приемом.
## Таблица ошибок
`ods.event_errors` держит строки, не прошедшие строгий приём, вместе с их сырым
текстом, метаданными доставки и классом брака. Ключи её собственные, потому что
у брака нет разобранных полей: шардируется `cityHash64` сырой строки —
`ClientID` у строки, которая не разобралась, взять неоткуда; нарезается по дню
загрузки, как и сырые данные; живёт месяц. Дольше сырых данных — намеренно: если
брак истекает вместе с ними, разбираться к моменту разбирательства будет уже
нечем. Снятие — целыми кусками (`ttl_only_drop_parts`), как у STG и по той же
причине: строки в партиции дня загрузки разного возраста не более чем на сутки,
и куску незачем переживать срок из-за самой свежей строки.
`ods.event_errors_dist` содержит сообщения, отвергнутые разбором событий:
исходный `raw`, сведения о доставке и класс ошибки `error_class`.
Класс имеет тип `LowCardinality(String)` и выбирается по первому
нарушенному условию:
Класс брака лежит в колонке `error_class` типа `LowCardinality(String)`. Без неё
в таблице копятся строки «что-то не так» без ответа на «что именно», а витрине
качества не на что опереться. Сами классы, их порядок и довод, почему порядок
обязателен, — в [ADR 0005](../adr/0005-event-ingestion.md).
1. `not_an_object` — корень JSON не объект;
2. `keyset_mismatch` — набор ключей не совпал с форматом;
3. `key_field_unparsed` — одно из пяти опорных полей не разобралось.
Движок — обычный `ReplicatedMergeTree`, без замены версий: схлопывать брак не по
чему, у него нет ключа сущности. `ORDER BY` — `(error_class, kafka_partition,
kafka_offset)`: смотрят такую таблицу от класса, а внутри класса — по координатам
доставки.
Порядок важен: строка вместо объекта нарушает сразу несколько условий,
но получает один класс. Полные условия — в
[ADR 0005](../adr/0005-event-ingestion.md).
`ods.order_errors` держит тот же диагностический минимум и `_load_id`
запуска. У заказов три класса по приоритету: `not_an_object`,
`keyset_mismatch`, `field_invalid`. Сырой текст остаётся рядом, поэтому класс не
разрастается до имени отдельного поля. Точная граница приёма — в
[спецификации заказов](orders/ingestion.md).
`ods.order_errors_dist` хранит такой же набор сведений и `_load_id`
запуска приема. Первые два класса совпадают с событиями,
третий называется `field_invalid`: у заказа проверяются все поля
верхнего уровня. Точные правила — в [описании приема заказов](orders/ingestion.md#граница-строгого-приёма).
Таблицы ошибок используют `ReplicatedMergeTree` без замены версий:
у некорректного сообщения может не быть разобранного ключа сущности.
Шард выбирается по `cityHash64(raw)`, ключ сортировки —
`(error_class, kafka_partition, kafka_offset)`, партиция — день `_load_ts`.
Начать поиск можно с класса ошибки, затем перейти к сообщениям
по координатам Kafka.
Срок хранения — один месяц от `_load_ts`, удаление целыми частями
с `ttl_only_drop_parts = 1`, как в STG. Ошибка вместе с исходным текстом
остается доступна дольше остальных сырых сообщений. Повторная обработка
может добавить тот же брак еще раз; у заказов повтор виден по совпадению
`_load_id` и координат Kafka.
## Справочники