# Перепишите разбор заказов в 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 находятся в зеленом состоянии.