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

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

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

Проблема

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

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

Цели

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

Не входит

• модель заказа в 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 полей проверяются только пять опорных; переносить то сокращение на малый контракт заказа нет причины. Граница строгости заканчивается на форме провода: содержимое позиций и бизнес-инварианты намеренно остаются ниже.

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

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

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

Роль ODS

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

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

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

Версию выбирает только updated_at. Заказ, не менявшийся несколько дней, приезжает в нескольких слепках одинаковыми деловыми полями, и какая из этих строк переживёт схлопывание — не обещано. Поэтому snapshot_date выжившей строки — один из дней, когда источник показывал эту версию, не обязательно последний; то же с _load_id и _load_ts. Вопрос «когда заказ наблюдался в последний раз» задаётся сырью STG, пока то живо, — а не одной строке ODS.

Сама схема версий стоит на допущении об источнике: уточнение заказа поднимает 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 вернёт короткую порцию, непрочитанный хвост останется в топике и приедет в один из следующих запусков. После отказа от замены партиции это задержка, а не потеря или публикация неполного дня.

Отказ — случай другой, и «хвост дождётся» на него не распространяется. Офсеты порции коммитятся в момент чтения, поэтому упавшая вставка уносит прочитанное с собой: повторное чтение вернёт ноль, а часть строк может уже лежать на шарде. Позиция мира при этом не двигается, и тот же день уедет заново — в сырье он окажется частичным дублем, а старый хвост, ушедший той же порцией, не вернётся (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. Опыты этого раздела прогнаны и подтвердили обещанное: разбиение сырья на годные строки и брак полное и непересекающееся; две версии одного заказа легли в одну партицию и на один шард, а 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 деловые поля совпали побайтово, по слепкам метка ни разу не убывала, а ods.order_v сошёлся построчно с эталоном argMax по updated_at, посчитанным прямо из сырья. Произвол ярлыков виден въявь: у 86 заказов стартового мира выжила строка серединного слепка — не первого и не последнего.

Забор из 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.