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

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

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

## Подготовка

- [проверки заказов](5-1-order-checks.md);
- [разбор заказов в ODS](5-2-order-parsing.md);
- [модель заказа DDS](../architecture/dds/order.md);
- [дневной конвейер](../architecture/etl.md#объём-прогона);
- [раскладку по шардам](../architecture/storage.md#раскладка-по-шардам).

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

Удалите готовый 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`. Контрольная сумма позволит проверить,
что после восстановления значения остались прежними. Удалите только
партицию выбранного дня:

```sql
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` поля заказов остались прежними, проверки прошли.