clickstream-data-platform/docs/labs/5-3-order-pipeline.md

Сборка заказов в DDS

Вы перепишете загрузку заказов из ODS в DDS и выбор дней для пересчета. Затем удалите данные за один из прошлых дней и проверите, что ваша загрузка восстановила их.

Подготовка

• проверки заказов;
• разбор заказов в ODS;
• модель заказа DDS;
• дневной конвейер;
• раскладку по шардам.

Файлы для изменения

Удалите готовый SQL и перепишите только три файла:

• sql/dds/order_scope.sql;
• sql/dds/order_rebuild.sql;
• sql/dds/order_replace.sql.

Даг dds_transform, DDL, промежуточную таблицу, ODS и обе проверки из первого задания оставьте как есть. Перед началом убедитесь, что пробные строки предыдущего задания удалены, а семь проверок качества дают зеленый результат.

Поставьте world_live на паузу и дождитесь завершения активных прогонов. До завершения опыта не запускайте world_live_day, world_initialize и world_recreate. world_next_day понадобится только в явно указанном ниже случае.

Выполнение запросов в Airflow

Даг dds_transform ожидает от файлов следующие параметры и результаты:

Файл	Вход	Результат
order_scope.sql	без параметров	одна строка first_day, last_day
order_rebuild.sql	{first_day:Date}, {last_day:Date}, {load_id:String}	выбранные дни в dds.order_stage_dist
order_replace.sql	день в {{ params.day }}	партиция дня в dds.order_rep

Сначала определите дни, за которые заказы еще могут измениться. Возьмите строки с последним snapshot_date и найдите минимальную и максимальную дату заказа. Этот диапазон далее называется текущим окном источника.

Затем найдите первый день, который есть в ODS, но отсутствует в DDS. Если он раньше начала текущего окна, начните расчет с него. Иначе начните с начала окна. Полученная дата станет first_day, а конец окна last_day. Один запуск пересчитывает все дни между ними. Если источник пуст, order_scope.sql не возвращает строку.

order_rebuild.sql сначала очищает все партиции dds.order_stage_rep на кластере, затем записывает в dds.order_stage_dist весь выбранный отрезок. Он читает текущее состояние ods.order_v, вычисляет order_date, переводит времена в Europe/Samara, разбирает items в три параллельных массива и заполняет служебные поля запуска.

Дождитесь записи всех строк на шарды через Distributed перед заменой партиций. Строки должны распределяться по ключу целевой таблицы. Способ рассчитать выбранные дни внутри файла выберите сами.

order_replace.sql работает с локальными dds.order_rep и dds.order_stage_rep на всем кластере. REPLACE PARTITION заменяет дневную партицию основной таблицы готовой партицией из промежуточной. Промежуточная партиция при этом сохраняется. Структура, ключи и политика хранения таблиц совпадают: DDL создает промежуточную через AS dds.order_rep.

Восстановление пропущенного дня

Найдите самый поздний день DDS перед началом текущего окна ODS. Заказы за этот день уже не меняются, поэтому он считается закрытым. Удалив его, вы проверите, что ваш запрос заметил пропуск и включил этот день в расчет.

Если такого дня еще нет, один раз продвиньте мир через world_next_day, дождитесь orders_ingest, затем запустите etl_pipeline. После этого снова оставьте модельное время неизменным до конца опыта и выберите этот день.

Сохраните для него число заказов и контрольную сумму их полей, исключив служебные _load_id и _load_ts. Контрольная сумма позволит проверить, что после восстановления значения остались прежними. Удалите только партицию выбранного дня:

ALTER TABLE dds.order_rep ON CLUSTER clickstream_cluster
DROP PARTITION 'YYYY-MM-DD';

Сразу запустите dq_check, до восстановления партиции. Красный результат должны дать orders_vs_ods и purchase_vs_orders_vs_dds. revenue_daily_vs_dds тоже даст красный результат, если в партиции были оплаченные заказы. classes_vs_inventory останется зеленым: до запуска dm_transform эта проверка читает прежнюю витрину сверки.

Запустите etl_pipeline. В группе dds_transform.order значение first_day должно совпасть с удаленным днем, а задачи replace - покрыть непрерывный отрезок от него до конца текущего окна. first_day ищите в строке про объем пересборки в журнале задачи order.scope дага dds_transform. После успешного прогона число заказов и контрольная сумма выбранного дня должны совпасть с сохраненными.

Распределение по шардам и повторный запуск

Напишите отдельный запрос к dds.order_dist, который получает фактический узел каждой строки через hostName(), вычисляет ожидаемый по cityHash64(order_id) % 2 и считает несовпадения по каждому узлу. Остаток 0 соответствует clickhouse-01, остаток 1 - clickhouse-02. На обоих узлах число строк на чужом шарде должно быть нулевым. Проверки количества строк читают данные со всех шардов через Distributed и такую ошибку не обнаруживают.

Запустите etl_pipeline еще раз при неизменном модельном времени. Группа заказов снова должна пересобрать текущее изменяемое окно. Контрольные суммы полей заказов остаются прежними, а все семь проверок дают зеленый результат.

Завершение работы

После восстановления партиции и успешных проверок снимите world_live с паузы, если он работал до упражнения.

Результат работы

• вы переписали три файла sql/dds/order_*.sql, сохранив остальные части стенда;
• вы удалили ближайший к окну закрытый день и увидели ошибку проверки до восстановления;
• ваш расчет выбрал first_day с удаленного дня и заменил партиции всего отрезка до конца окна;
• вы сверили число заказов и контрольную сумму восстановленного дня с сохраненными значениями; они совпали;
• order_items_total, orders_vs_ods и пять прежних проверок успешны;
• ваш запрос с hostName() не нашел заказов на чужом шарде;
• после повторного etl_pipeline поля заказов остались прежними, проверки прошли.