clickstream-data-platform/docs/labs/5-2-order-parsing.md

# Перепишите разбор заказов в 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"
```

Отправьте файл в обычный топик заказов утилитой из контейнера Kafka:

```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,
временные файлы удалены, а семь проверок успешны. Теперь можно перейти к
сборке DDS, не снимая паузу с модельного времени. Если вы прерываетесь и до
опыта `world_live` работал, снимите его с паузы.

## Что должно получиться

- [ ] у вас сохранены результаты до переписывания: четыре строки дали
      один заказ и три класса брака; вы сверили поля текущей версии и шард;
- [ ] вы переписали `sql/ods/order_parse.sql`, сохранив остальные части стенда;
- [ ] ваш разбор дал тот же результат на новой пробной пачке;
- [ ] вы проверили разные значения `created_at` и `updated_at` у годной
      строки в обоих опытах и подтвердили правильный шард через `hostName()`;
- [ ] вы удалили строки обоих опытов по их маркерам;
- [ ] после очистки конвейер и все семь проверок успешны.