docs/labs/5-1-order-checks.md

@@ -0,0 +1,148 @@
+# Сначала сторож, потом ломай: заказы
+
+**Чему учит:** до рискованного изменения задавать проверке явный договор и
+испытывать его управляемой ошибкой.
+
+## Сначала прочитать
+
+- [модель заказа DDS](../architecture/dds/order.md);
+- [дневной конвейер](../architecture/etl.md#объём-прогона);
+- [правила проверок](../architecture/testing.md#проверка-которая-не-умеет-краснеть-бесполезна).
+
+Посмотрите также на устройство одной проверки в [`sql/dq/`](../../sql/dq/)
+и на ее включение в [`dq_check`](../../dags/dq_check.py). Готового ответа для
+двух проверок из этого задания в репозитории нет.
+
+## Граница задания
+
+Добавьте два сторожа:
+
+- `sql/dq/order_items_total.sql` проверяет сумму позиций заказа;
+- `sql/dq/orders_vs_ods.sql` сверяет DDS с его источником;
+- `dags/dq_check.py` запускает обе проверки, публикует их строки в
+  `dm.dq_summary` и создает для них отдельные задачи.
+
+На этом шаге не меняйте разбор ODS и сборку DDS. Даги, DDL, таблицы-доноры,
+Kafka и службы остаются несущей частью упражнения.
+
+Перед опытами остановите автоматическое движение мира: поставьте `world_live`
+на паузу и дождитесь завершения активных прогонов. До чистой точки остановки не
+запускайте руками `world_next_day`, `world_live_day`, `world_initialize` и
+`world_recreate`. Это условие стабильного модельного времени: одной паузы
+`world_live` для него мало.
+
+## Два договора качества
+
+Обе проверки следуют договору `dq_check`: запрос возвращает по одной строке
+на день с колонками `data_date`, `reference_rows`, `actual_rows`,
+`failed_rows`, `diagnostics`. Даг сохраняет в `dm.dq_summary` только счетчики,
+статус и координаты запуска. Ограниченные примеры из `diagnostics` ищите в
+журнале упавшей задачи Airflow. Ограничьте их тем же малым числом, что и
+существующие запросы в `sql/dq/`.
+
+### Сумма позиций
+
+Имя проверки - `order_items_total`. Для каждого заказа из `dds.order_v`
+посчитайте сумму `item_price * item_quantity` и сравните ее с `items_total`.
+`reference_rows` считает ключи со стороны `items_total`, `actual_rows` - ключи
+со стороны вычисленной суммы, а `failed_rows` - несовпавшие значения. В
+исправном наборе первые два счетчика равны по построению; сам инвариант выражен
+через `failed_rows`. Для диагностики достаточно `order_id`, ожидаемой суммы и
+суммы позиций.
+
+Проверка относится к типизированному разбору массива `items` на границе
+ODS -> DDS. Скидка, доставка и равенство `total` в этот договор не входят.
+
+### DDS против ODS
+
+Имя проверки - `orders_vs_ods`. Источник модели DDS здесь - текущее состояние
+`ods.order_v`. Внутреннее состояние генератора в этот договор не входит.
+Одним запросом соберите обе стороны на зерне дня заказа и сравните:
+
+- число заказов;
+- максимальный `updated_at`.
+
+День ODS вычисляется в `Europe/Samara`, как `order_date` в DDS. Область
+проверки - объединение дней обеих сторон, включая день, который полностью
+пропал из одного слоя. Перед сравнением приведите времена к одному часовому
+поясу. `reference_rows` - число заказов ODS, `actual_rows` - число заказов DDS,
+а `failed_rows` равен единице для дня, где расходится хотя бы одно значение.
+
+Обе стороны распределены по `order_id`. Дневные агрегаты соедините через
+`GLOBAL FULL OUTER JOIN` с `join_use_nulls = 1`, как в существующих сверках.
+
+Если разбор ODS и сборка DDS одинаково потеряют строку, слои останутся равны
+по этой проверке. Общая потеря при разборе проявится в отдельном балансе по
+`_load_id` из [следующего задания](5-2-order-parsing.md).
+
+## Получите красный и зеленый результаты
+
+Сначала запустите `dq_check` на нетронутых данных. Обе новые задачи должны
+завершиться успешно, а в `dm.dq_summary_v` у обоих имен не должно быть
+нарушенных дней.
+
+Выберите оплаченный заказ из последнего слепка с непустым массивом позиций и
+сохраните его `order_id` и `order_date`. Подключитесь к первой ноде из DBeaver
+по [инструкции из README](../../README.md#подключиться-из-dbeaver) под
+пользователем `etl`. Прибавьте одну копейку к `items_total` этого заказа в
+локальной таблице `dds.order_rep` на всем кластере:
+
+```sql
+ALTER TABLE dds.order_rep ON CLUSTER clickstream_cluster
+UPDATE items_total = items_total + toDecimal64(0.01, 2)
+WHERE order_id = '<order_id>'
+SETTINGS mutations_sync = 2;
+```
+
+Затем запустите только `dq_check`. Задача `order_items_total` должна
+завершиться ошибкой, а ее журнал - назвать выбранный заказ.
+`purchase_vs_orders_vs_dds` тоже даст красный результат: сохраненная витрина
+содержит прежний `items_total`.
+
+Запустите `etl_pipeline`. Текущее окно заказов будет заново собрано из ODS, и
+исходная строка вернется в DDS. Повторный `dq_check` должен дать зеленый
+результат для нового сторожа.
+
+Теперь удалите тот же заказ из `dds.order_rep` на всем кластере:
+
+```sql
+ALTER TABLE dds.order_rep ON CLUSTER clickstream_cluster
+DELETE WHERE order_id = '<order_id>'
+SETTINGS mutations_sync = 2;
+```
+
+До восстановления снова запустите только `dq_check`. Красный результат нужен
+для `orders_vs_ods`, `purchase_vs_orders_vs_dds` и `revenue_daily_vs_dds` на
+дне оплаченного заказа. Затем запустите `etl_pipeline` и повторите проверку.
+Если `orders_vs_ods` не обнаружил удаление, исправьте запрос до следующего
+задания.
+
+## Как проверить себя
+
+В `dm.dq_summary_v` должны быть зелены обе новые и пять прежних проверок:
+
+```sql
+SELECT check_name, sum(failed_rows) AS failed_rows, groupUniqArray(status)
+FROM dm.dq_summary_v
+GROUP BY check_name
+ORDER BY check_name;
+```
+
+Проверка суммы должна охватывать каждый день `dds.order_v`. Сверка слоев
+должна охватывать объединение дней ODS и DDS, включая дни только одной стороны.
+
+## Когда можно остановиться
+
+После двух восстановлений выбранный заказ снова совпадает с ODS. Все семь
+проверок дают зеленый результат. В таком состоянии можно закончить работу или
+перейти к разбору заказов, оставив модельное время на паузе.
+
+## Приемка
+
+- [ ] две новые проверки встроены в `dq_check` и публикуют дневные строки по
+      принятому договору;
+- [ ] каждая проверка дала красный результат на своей управляемой ошибке;
+- [ ] журналы упавших задач содержали ограниченную диагностику;
+- [ ] `etl_pipeline` восстановил обе измененные строки DDS;
+- [ ] `order_items_total`, `orders_vs_ods` и пять прежних DQ зелены;
+- [ ] SQL разбора ODS и сборки DDS на этом шаге не менялся.

docs/labs/5-2-order-parsing.md

@@ -0,0 +1,181 @@
+# Перепишите разбор заказов в ODS
+
+**Чему учит:** сначала измерить обе ветви разбора на управляемом срезе, затем
+переписать переход и повторить тот же опыт.
+
+## Сначала прочитать
+
+- [проверки заказов](5-1-order-checks.md);
+- [прием заказов в ODS](../architecture/orders/ingestion.md);
+- [раскладку по шардам](../architecture/storage.md#раскладка-по-шардам);
+- [`orders_ingest`](../../dags/orders_ingest.py);
+- [`order_parse.sql`](../../sql/ods/order_parse.sql).
+
+Готового запроса для сверки управляемого среза в репозитории нет.
+
+## Граница задания
+
+Сначала проведите опыт на готовом `order_parse.sql`. Затем удалите его SQL,
+соберите разбор заново и повторите опыт с другим маркером.
+
+Не меняйте устройство `orders_ingest`, DDL, Kafka, таблиц и преобразований
+DDS. Оба опыта должны закончиться адресной очисткой только своих строк.
+
+Перед началом убедитесь, что после [предыдущего задания](5-1-order-checks.md)
+все семь DQ остались зелеными. Поставьте `world_live` на паузу, дождитесь
+активных прогонов и до завершения опыта не запускайте `world_next_day`,
+`world_live_day`, `world_initialize` и `world_recreate`. Не запускайте
+`etl_pipeline` между
+отправкой пробной пачки и ее очисткой.
+
+## Четыре строки для двух ветвей
+
+Сделайте четыре сообщения с общим уникальным маркером:
+
+1. годную строку с маркером в `order_id`;
+2. JSON-строку вместо объекта - класс `not_an_object`;
+3. объект с лишним ключом - класс `keyset_mismatch`;
+4. объект с точным набором ключей, но строкой в `user_id` - класс
+   `field_invalid`.
+
+`not_an_object` означает, что корень JSON не объект. `keyset_mismatch` означает,
+что у объекта отличается точный набор верхнеуровневых ключей. `field_invalid`
+означает, что ключи на месте, но значение не прошло проверку типа, формата или
+вместимости.
+
+Соберите строки в `/tmp/<маркер>.jsonl`. Заготовка дает одну годную строку и
+по одному примеру каждого класса брака:
+
+```bash
+LAB_MARKER="lab5-$(date +%s)"
+LAB_BATCH="/tmp/${LAB_MARKER}.jsonl"
+LAB_DAY="$(
+  docker compose exec -T clickhouse-01 sh -lc \
+    'clickhouse-client --user etl --password "$CLICKHOUSE_ETL_PASSWORD" \
+      --query "SELECT toString(max(snapshot_date)) FROM ods.order_dist"'
+)"
+
+jq -cn --arg marker "$LAB_MARKER" --arg day "$LAB_DAY" '
+  def good($id): {
+    order_id: $id, user_id: 1, status: "paid",
+    created_at: ($day + "T00:00:00.000Z"),
+    updated_at: ($day + "T01:00:00.000Z"),
+    items_total: "10.00", discount: "0.00",
+    delivery: "0.00", total: "10.00",
+    items: [{sku: "lab5", qty: 1, price: "10.00"}],
+    snapshot_date: $day
+  };
+  good($marker),
+  ($marker + "-not-an-object"),
+  (good($marker + "-keys") + {lab_marker: true}),
+  (good($marker + "-field") | .user_id = "broken")
+' > "$LAB_BATCH"
+```
+
+Отправьте файл в обычный топик заказов штатным производителем контейнера:
+
+```bash
+docker compose exec -T kafka /opt/kafka/bin/kafka-console-producer.sh \
+  --bootstrap-server kafka:9092 --topic orders < "$LAB_BATCH"
+```
+
+Запустите `orders_ingest` и найдите `_load_id` по маркеру в
+`stg.orders_raw_dist`. У четырех строк должен быть один идентификатор прогона.
+
+## Зафиксируйте исходный результат
+
+До любого запуска `etl_pipeline` проверьте срез по найденному `_load_id`:
+
+- в STG четыре строки;
+- в физической `ods.order_dist` одна годная строка;
+- в `ods.order_errors_dist` три строки, по одной каждого класса;
+- `STG = годные + брак`;
+- текущая строка `ods.order_v` совпадает по деловым полям с результатом
+  `argMax` по `snapshot_date` из годного сырья среза.
+
+Счет физической таблицы нужен именно по `_load_id`: представление
+`ods.order_v` оставляет текущую версию сущности и не показывает, сколько строк
+разобрал запуск. Запрос сверки с `argMax` напишите сами. Эталонные деловые поля
+стройте только из `raw`; отдельно сравните различающиеся `created_at` и
+`updated_at`.
+
+Проверьте раскладку годной строки через `hostName()` на `ods.order_dist`.
+Фактическая нода должна совпасть с остатком `cityHash64(order_id) % 2`:
+0 - `clickhouse-01`, 1 - `clickhouse-02`.
+
+## Уберите пробную пачку
+
+Удалите только строки маркера из локальных таблиц:
+
+```sql
+ALTER TABLE stg.orders_raw_rep ON CLUSTER clickstream_cluster
+DELETE WHERE position(raw, '<маркер>') > 0
+SETTINGS mutations_sync = 2;
+
+ALTER TABLE ods.order_rep ON CLUSTER clickstream_cluster
+DELETE WHERE startsWith(order_id, '<маркер>')
+SETTINGS mutations_sync = 2;
+
+ALTER TABLE ods.order_errors_rep ON CLUSTER clickstream_cluster
+DELETE WHERE position(raw, '<маркер>') > 0
+SETTINGS mutations_sync = 2;
+```
+
+В ClickHouse 26.3 мутации выполняются в фоне. С `mutations_sync = 2` команда
+ждет их завершения на всех репликах. Запросами к трем Distributed-таблицам
+докажите, что строк маркера не осталось, затем удалите временный файл.
+
+Группа `clickstream_orders` уже подтвердила эти сообщения, поэтому обычный
+повтор `orders_ingest` их не прочитает. В Kafka они остаются до удаления по
+политике хранения. `world_recreate` удаляет группу и сам топик, поэтому в
+пересозданный мир пробная пачка не переходит.
+
+Очистка должна закончиться до запуска `etl_pipeline`. Если конвейер все же
+успел собрать временный заказ в DDS, сначала очистите три таблицы выше, затем
+запустите `etl_pipeline`: пересборка текущего окна удалит заказ из DDS и
+обновит витрины.
+
+## Перепишите разбор
+
+Удалите готовый SQL из `sql/ods/order_parse.sql` и сохраните его договор с
+`orders_ingest`:
+
+- вход - строки `stg.orders_raw_dist` одного `{load_id:String}`;
+- каждая строка среза попадает ровно в один результат: годный заказ или брак;
+- произвольный `raw` не останавливает весь разбор исключением;
+- годные строки записываются в `ods.order_dist`, остальные - в
+  `ods.order_errors_dist`;
+- классы брака и их порядок - `not_an_object`, `keyset_mismatch`,
+  `field_invalid`;
+- метки доставки и `_load_id` переходят из STG без пересчета;
+- задача завершается после доставки обоих результатов через Distributed, а
+  строки распределяются по ключу целевой таблицы.
+
+Точный набор полей, типы и граница строгого приема описаны в
+[документе приема](../architecture/orders/ingestion.md#граница-строгого-приёма),
+а целевые колонки - в [`20-ods-tables.sql`](../../sql/ddl/20-ods-tables.sql).
+Внутренность `items` остается сырым JSON: ее разбирает DDS.
+
+Повторите опыт с новым маркером. Получите тот же баланс 4 = 1 + 3, те же три
+класса, верную текущую строку по `snapshot_date`, разные времена и правильный
+шард. Затем очистите второй срез теми же адресными командами и запустите
+`dq_check`.
+
+## Когда можно остановиться
+
+Оба маркера отсутствуют в STG, ODS, таблице брака и DDS; временные файлы
+удалены. Семь DQ дают зеленый результат. В таком состоянии можно перейти к
+сборке DDS, не снимая паузу с модельного времени. Если вы прерываетесь и до
+опыта `world_live` работал, снимите его с паузы.
+
+## Приемка
+
+- [ ] до переписывания зафиксированы баланс 4 = 1 + 3, три класса брака,
+      деловые поля текущей версии и фактический шард;
+- [ ] переписан только `sql/ods/order_parse.sql`, несущая инфраструктура не
+      изменена;
+- [ ] после переписывания новый срез дал тот же результат;
+- [ ] обе годные строки различали `created_at` и `updated_at`;
+- [ ] проверки `hostName()` не нашли строки ODS на чужом шарде;
+- [ ] строки обоих опытов удалены только по своим маркерам;
+- [ ] после очистки общий конвейер и семь DQ находятся в зеленом состоянии.

docs/labs/5-3-order-pipeline.md

@@ -0,0 +1,124 @@
+# Перепишите сборку заказов в 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 и обе проверки из первого задания уже
+задают границу работы. Перед началом убедитесь, что пробные строки предыдущего
+задания удалены, а семь DQ дают зеленый результат.
+
+Поставьте `world_live` на паузу и дождитесь завершения активных прогонов. До
+завершения опыта не запускайте `world_live_day`, `world_initialize` и
+`world_recreate`. `world_next_day` понадобится только в явно указанном ниже
+случае.
+
+## Договор с дагом
+
+Сохраните интерфейс `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` |
+
+Объем состоит из текущего окна источника и хвоста от первого дня, который есть
+в ODS, но отсутствует в DDS. Текущее окно дают строки последнего
+`snapshot_date`: их минимальный и максимальный день заказа. Если источник
+пуст, `order_scope.sql` не возвращает строку. Если дыр нет, `first_day` равен
+началу окна. Если дыра старше, непрерывный отрезок начинается с нее.
+
+`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`. На обеих нодах число строк на
+чужом шарде должно быть нулевым. Счетные DQ читают Distributed-таблицы и такую
+ошибку раскладки не обнаруживают.
+
+Запустите `etl_pipeline` еще раз при неизменном модельном времени. Группа
+заказов снова должна пересобрать текущее изменяемое окно. Контрольные суммы
+деловых полей остаются прежними, а все семь проверок дают зеленый результат.
+
+## Когда можно остановиться
+
+Удаленная партиция восстановлена, ее число строк и контрольная сумма совпали
+с исходными. На чужих шардах строк нет, повторный прогон не изменил деловые
+данные, семь DQ зелены. Если до упражнения `world_live` работал, теперь его
+можно снять с паузы.
+
+## Приемка
+
+- [ ] переписаны только три файла `sql/dds/order_*.sql`, инфраструктура
+      упражнения не изменена;
+- [ ] удален ближайший к окну закрытый день, а красный DQ зафиксирован до
+      восстановления;
+- [ ] `first_day` начался с удаленного дня, задачи замены покрыли непрерывный
+      отрезок до конца окна;
+- [ ] число заказов и контрольная сумма восстановленного дня совпали;
+- [ ] `order_items_total`, `orders_vs_ods` и пять прежних DQ зелены;
+- [ ] проверка `hostName()` не нашла заказов DDS на чужом шарде;
+- [ ] повторный `etl_pipeline` сохранил деловые данные и зеленые проверки.

docs/labs/README.md

@@ -12,11 +12,11 @@
 | 2 | [Свой график в Superset](2-1-superset-chart.md) | превратить свой результат из хранилища в отчет в BI |
 | 3 | [Нарежь сессии сам](3-1-sessions-by-hand.md) | сверить число сессий в своей нарезке с эталоном и найти события, из-за которых оно расходится |
 | 4 | [Перепиши нарезку стенда](4-1-session-pipeline.md) | изменить расчетную сущность внутри готового конвейера и проверить ее внешним эталоном |
-| 5 | Сначала сторож, потом ломай: заказы | написать проверки качества до рискованного изменения, а не после первой поломки |
+| 5 | [Сначала сторож, потом ломай: заказы](5-1-order-checks.md)<br>[Перепишите разбор заказов в ODS](5-2-order-parsing.md)<br>[Перепишите сборку заказов в DDS](5-3-order-pipeline.md) | написать проверки качества до рискованного изменения, затем переписать разбор и сборку заказов |
 | 6 | Разбери сырые события и сверь результат со стендом | изучить разбор потока отдельно от живого приема и сверить его со стендовым результатом |
 | 7 | Мониторинг: куда смотреть | выбрать, куда и на что смотреть в живой системе |
 
-ТЗ для ступеней 5-7 появятся по мере продвижения программы.
+ТЗ для ступеней 6-7 появятся по мере продвижения программы.
 
 ## Ступень 1: пять витрин слоя DM