clickstream-data-platform/docs/architecture/orders/ingestion.md

Приём заказов из Kafka в ODS

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

Проблема

Заказы приезжают полным слепком окна изменяемости, но Kafka передаёт его отдельными сообщениями и не сообщает потребителю, где слепок закончился. Прямое чтение Kafka Engine возвращает одну порцию без признака полноты. Поэтому snapshot_date не может служить границей публикации: REPLACE PARTITION могла бы заменить партицию дня неполным набором строк.

Одновременно типизированный ODS не должен принимать правдоподобные значения по умолчанию из грязного JSON или останавливать весь пакет из-за одной строки.

Цели

• сохранить пакетный забор как контраст потоковому приёму событий;
• один раз принять байты в STG и независимо разложить строки на годные и брак;
• хранить в ODS типизированные версии заказов, не публикуя слепок заменой партиции;
• дать следующим слоям один корректный способ прочитать текущее состояние;
• оставить код проверки коротким и ограничить его контрактом провода.

Не входит

• модель заказа в DDS: её зерно, связи, материализация и способ наполнения;
• проверка полей внутри items, переходов статуса и равенств денежных сумм;
• удаление заказа по отсутствию в следующем слепке;
• маркер конца слепка, опись ожидаемых строк и транзакция между целями ODS.

Поток данных

После завершения генератора Airflow один раз читает байтовый Kafka-чтец и записывает полученную порцию в stg.orders_raw. Даг зовётся orders_ingest, а запускает его шаговый даг модельного времени, который положил слепок в топик и ждёт конца прогона. Все строки получают _load_id, равный run_id Airflow. _load_ts вычисляется при этой записи и дальше переносится без пересчёта. Запрос забора лежит в sql/stg/orders_raw_load.sql: SQLExecuteQueryOperator задаёт подключение и передаёт run_id как параметр ClickHouse load_id, а преобразование остаётся в SQL своего слоя.

Один следующий task_id отвечает за весь переход STG → ODS. Внутри него два последовательных INSERT SELECT читают неизменный срез по _load_id: первый пишет годные строки в ods.order, второй — брак в ods.order_errors. Транзакции между запросами нет. При частичном сбое Airflow повторяет весь task_id; одинаковые исходные строки и служебные метки не вычисляются заново. Обе вставки лежат в одном самодостаточном файле order_parse.sql. Airflow сначала шаблонизирует файл, затем делит получившийся SQL на выражения и исполняет их по порядку в одной задаче parse_batch.

Условия запросов взаимодополняющие: один общий предикат определяет годную строку, а ветвь брака использует его буквальное отрицание. Все функции предиката возвращают результат без исключения, а сам предикат всегда заканчивается в true или false, не в NULL. Постоянный классификатор между STG и ODS для этого не нужен. Предикат объявлен локальным макросом Jinja внутри order_parse.sql и разворачивается в оба выражения. Макрос убирает дублирование, но не вводит отдельный фрагмент, который нельзя исполнить самостоятельно.

Граница строгого приёма

Единица решения — одна строка stg.orders_raw. Корень должен быть 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-массив. Каноническая форма и основания выбора зафиксированы в исследовании формата.

Проверять все верхнеуровневые поля здесь уместно: их одиннадцать, и десять скалярных значений непосредственно образуют типизированную строку заказа. У события из 47 полей проверяются только пять опорных; переносить то сокращение на малый контракт заказа нет причины. Граница строгости заканчивается на форме провода: содержимое позиций и бизнес-инварианты намеренно остаются ниже.

Контракт источника задаёт частоту: одной дате соответствует один слепок. Это свойство всего слепка, поэтому предикат одной строки его не проверяет. Вторая выгрузка того же дня даст ничью по snapshot_date, и выбор строки потеряет однозначность. Модельный источник выполняет условие; в настоящей интеграции его фиксируют в договоре с источником (ADR 0014).

Класс брака выбирается первым совпадением:

1. not_an_object;
2. keyset_mismatch;
3. field_invalid.

Имя отдельного поля в класс не включается. В таблице ошибок остаются сырой текст, метаданные доставки и _load_id, поэтому единичный случай можно разобрать без постоянной детализации предиката.

Роль ODS

ods.order_rep хранит физически принятые версии в ReplacingMergeTree(snapshot_date). Ключ сортировки — order_id, партиция — день неизменного created_at. ods.order_dist шардирует по cityHash64(order_id): только так все версии заказа попадают на один шард и FINAL даёт корректный результат через распределённую таблицу.

Четыре координаты отвечают на разные вопросы:

• snapshot_date — в слепке какого модельного дня источник показал строку;
• updated_at — когда строку последний раз меняли в источнике;
• _load_id — какой запуск Airflow принял строку;
• _load_ts — когда строка приехала в хранилище.

Версию выбирает snapshot_date: выживает строка с максимальной датой слепка (ADR 0014). Поэтому snapshot_date показывает последнее наблюдение заказа. По строкам последней даты слепка DDS определяет окно пересборки. _load_id и _load_ts принадлежат той физической строке, которую сохранило слияние, и описывают её доставку.

Схема версий стоит на дисциплине источника: слепок с поздней датой не показывает состояния старше. Модельный мир её держит — состояние читается на границе суток, и слепок дня выходит чистой функцией зерна и дня; источник, собирающий слепок из отставшей реплики, её ломает. Деловая метка updated_at при этом монотонна по слепкам, поэтому выбор по дате слепка даёт тот же ответ, что выбор по метке (замер — «Что проверено»). Тихая правка строки — ручным UPDATE мимо приложения, миграцией — метку не двигает, но следующий слепок перекроет её сам, пока заказ в окне изменяемости. За окном источник заказ больше не показывает, и там помогает только сверка самих полей, в терминах dbt snapshots — стратегия check вместо timestamp.

Обычное чтение _dist показывает физически сохранившиеся версии и нужно для диагностики. Их число зависит от фоновых слияний: ODS не служит архивом истории. ods.order_v сохраняет те же источник-ориентированные поля без обогащения данными модели, но возвращает одну актуальную версию на order_id. Сначала оно может быть простым представлением над _dist FINAL; способ выбора можно заменить, не меняя потребителей.

Это представление остаётся ответственностью ODS: оно скрывает механику чтения версий, но не строит бизнес-модель. DDS читает ods.order_v и отдельно решает, какие сущности, связи и производные признаки ему нужны. Судьба служебных колонок выше ODS решена вместе с моделью DDS: чужие ярлыки туда не переезжают (модель заказа).

Одно чтение Kafka

Стандартный слепок содержит около полутора тысяч строк, тогда как предел одной порции на стенде — десятки тысяч сообщений. Поэтому один запуск Airflow делает один прямой SELECT, без цикла до пустоты и без фиксации конечных офсетов.

Это допущение о размере стенда, а не доказательство полноты слепка. Если Kafka вернёт короткую порцию, непрочитанный хвост останется в топике и приедет в один из следующих запусков. После отказа от замены партиции это задержка, а не потеря или публикация неполного дня.

Отказ — случай другой, и «хвост дождётся» на него не распространяется. Офсеты порции коммитятся в момент чтения, поэтому упавшая вставка уносит прочитанное с собой: повторное чтение вернёт ноль, а часть строк может уже лежать на шарде. Позиция мира при этом не двигается, и тот же день уедет заново — в STG он окажется частичным дублем, а старый хвост, ушедший той же порцией, не вернётся (ADR 0008, «Следствия»).

Отклонённые варианты

• Партиционная идемпотентность — 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-координаты показывают повтор задачи.

Что проверено

Переход STG → ODS снят на живом стенде 18 августа 2026 года при исполнении #94. Опыты этого раздела прогнаны и подтвердили обещанное: разбиение строк STG на годные строки и брак полное и непересекающееся; две версии одного заказа легли в одну партицию и на один шард, а ods.order_v вернуло позднюю независимо от фонового слияния; повтор задачи с тем же _load_id строку в ods.order_v и её _load_ts не изменил, а таблица ошибок записала тот же брак второй раз; строки опытов убраны. Числа, механика опытов и поведение ClickHouse, на которое всё это опирается, — документ хранилища, «Что проверено».

После исправления #102 честный прогон дня снят на живом стенде 20 августа 2026 года: все 1694 строки слепка дня 7 приняты, ods.order_errors осталась пустой. У заказа 20260608-0007, который прежде приходил с отрицательным итогом, теперь 5490.00 − 823.50 + 299.00 = 4965.50.

При исполнении #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, раздел «Что проверено».

MCP Context7 в сессии проектирования был недоступен. На локальном ClickHouse 26.3.17.56 проверено, что прямой SELECT Kafka Engine завершается после одной порции, а FINAL через Distributed исполняется на таблицах шардов. Поэтому версии одного order_id направляются на один шард. Фоновое схлопывание ReplacingMergeTree и необходимость точного чтения сверены с официальной документацией, поведение чтения — с исходниками той же версии StorageKafka.cpp и KafkaSource.cpp.

Связанные решения

• ADR 0008 сохраняет выбор пакетного забора, RawBLOB, одного чтеца и одной партиции топика.
• ADR 0010 заменяет публикацию слепка версионным ODS.