From d686de516b3eb6b63a45748becf0503d3c4744be Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Sun, 6 Sep 2026 15:12:11 -0400 Subject: [PATCH 1/2] =?UTF-8?q?docs(labs):=20=D0=B4=D0=BE=D0=B1=D0=B0?= =?UTF-8?q?=D0=B2=D0=BB=D0=B5=D0=BD=D0=BE=20=D0=B7=D0=B0=D0=B4=D0=B0=D0=BD?= =?UTF-8?q?=D0=B8=D0=B5=20=D0=BF=D0=BE=20=D0=BF=D1=80=D0=BE=D0=B2=D0=B5?= =?UTF-8?q?=D1=80=D0=BA=D0=B0=D0=BC=20=D0=B7=D0=B0=D0=BA=D0=B0=D0=B7=D0=BE?= =?UTF-8?q?=D0=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Зачем: - менти нужен наблюдаемый опыт, в котором проверки защищают изменение конвейера заказов. - Что: - добавлено ТЗ ступени 5 с договорами DQ, управляемыми ошибками и восстановлением. - в оглавление лабораторных работ добавлена ссылка на новое задание. - Проверка: - проверены локальные ссылки, пробная пачка jq и отсутствие ошибок пробелов в diff. --- docs/labs/5-1-order-checks.md | 347 ++++++++++++++++++++++++++++++++++ docs/labs/README.md | 4 +- 2 files changed, 349 insertions(+), 2 deletions(-) create mode 100644 docs/labs/5-1-order-checks.md diff --git a/docs/labs/5-1-order-checks.md b/docs/labs/5-1-order-checks.md new file mode 100644 index 0000000..2abfd85 --- /dev/null +++ b/docs/labs/5-1-order-checks.md @@ -0,0 +1,347 @@ +# Сначала сторож, потом ломай: заказы + +**Чему учит:** ставить проверку качества до рискованного изменения и +доказывать управляемой ошибкой, что она защищает нужный договор. + +## Сначала прочитать + +- [прием заказов в ODS](../architecture/orders/ingestion.md); +- [модель заказа DDS](../architecture/dds/order.md); +- [дневной конвейер](../architecture/etl.md#объём-прогона); +- [раскладку по шардам](../architecture/storage.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` и красит именованные задачи. + +Только после красного и зеленого опыта с каждым сторожем перепишите готовые +преобразования: + +- `sql/ods/order_parse.sql`; +- `sql/dds/order_scope.sql`; +- `sql/dds/order_rebuild.sql`; +- `sql/dds/order_replace.sql`. + +Даги приема и трансформации, DDL, таблицы-доноры, Kafka и службы остаются +несущей частью упражнения. Их интерфейсы уже заданы. Перед опытами остановите +автоматическое движение мира: поставьте `world_live` на паузу и дождитесь +завершения активных прогонов. До конца очередного опыта не запускайте +`world_next_day`, `world_live_day` и `world_initialize` руками. Это условие +стабильного модельного времени: одной паузы `world_live` для него мало. Даг +`world_recreate` тоже оставьте нетронутым: он пересоздает весь прикладной мир. + +## Два договора качества + +Обе проверки следуют договору `dq_check`: запрос возвращает по одной строке +на день с колонками `data_date`, `reference_rows`, `actual_rows`, +`failed_rows`, `diagnostics`. Даг записывает результат под отдельным +`check_name` и падает, если `failed_rows` больше нуля. Диагностика называет +день, `order_id` или сравниваемые дневные значения. Ограничьте ее тем же +малым числом примеров, что и существующие запросы. + +### Сумма позиций + +Имя проверки - `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`: его вы проверите до запуска нижележащих преобразований. + +## Докажите, что сторожа работают + +Сначала запустите `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 = '' +SETTINGS mutations_sync = 2; +``` + +Затем запустите только `dq_check`. + +`order_items_total` должен стать красным и назвать выбранный заказ. +`purchase_vs_orders_vs_dds` тоже покраснеет: его сохраненная витрина еще несет +прежний `items_total`. Исправьте опыт запуском `etl_pipeline`. Заказы +пересобирают всё текущее окно, и строка вернется из ODS. Убедитесь, что новый +сторож снова зеленый. + +Теперь удалите тот же заказ из `dds.order_rep` на всем кластере: + +```sql +ALTER TABLE dds.order_rep ON CLUSTER clickstream_cluster +DELETE WHERE order_id = '' +SETTINGS mutations_sync = 2; +``` + +До восстановления запустите только `dq_check`: `orders_vs_ods`, +`purchase_vs_orders_vs_dds` и `revenue_daily_vs_dds` должны покраснеть на дне +оплаченного заказа. Затем запустите `etl_pipeline` и получите зеленый +результат. Если новая проверка не увидела свою поломку, чините ее до перехода +к разбору. + +## Проверьте обе ветви разбора + +Чистый мир не производит брак заказов. Поэтому до вычитания +`order_parse.sql` проведите отдельный опыт, а после своей реализации повторите +его с новым маркером. + +Сделайте четыре сообщения с общим уникальным маркером: + +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` проверьте этот срез: + +- в 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-таблицам +докажите, что строк маркера не осталось, и удалите временный файл. Сообщения +в Kafka уже подтверждены группой `clickstream_orders` и повторно не читаются. +В журнале они останутся до удаления по политике хранения топика. +Очистка должна закончиться до любого запуска `etl_pipeline`. Если конвейер все +же успел собрать временный заказ в DDS, после очистки запустите его еще раз: +обычная пересборка текущего окна уберет строку маркера. + +## Перепишите разбор ODS + +Удалите готовый SQL из `sql/ods/order_parse.sql` и соберите переход заново. +Его интерфейс остается прежним: + +- вход - строки `stg.orders_raw_dist` одного `{load_id:String}`; +- первая вставка пишет годные строки в `ods.order_dist`; +- вторая пишет остальные строки в `ods.order_errors_dist`; +- классы брака и их порядок - `not_an_object`, `keyset_mismatch`, + `field_invalid`; +- обе ветви используют один предикат, который на произвольном `raw` дает + только 0 или 1 и не бросает исключение; +- метки доставки и `_load_id` переходят из STG без пересчета; +- обе вставки дожидаются Distributed-записи и отдают раскладку ключу цели. + +Точный набор полей, типы и граница строгого приема описаны в +[документе приема](../architecture/orders/ingestion.md#граница-строгого-приёма), +а целевые колонки - в [`20-ods-tables.sql`](../../sql/ddl/20-ods-tables.sql). +Внутренность `items` на этой границе остается сырым JSON: ее разбирает DDS. + +Повторите опыт четырех строк с новым маркером. Сначала получите тот же баланс, +три класса, текущую версию по `snapshot_date`, разные времена и правильный +шард. Только затем очистите срез и переходите в DDS. + +## Перепишите модель DDS + +Удалите готовый SQL из трех файлов `sql/dds/order_*.sql` и сохраните договор с +`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` равен +началу окна. Если дыра старше, он расширяет непрерывный отрезок влево до нее. + +Пересборка сначала очищает все партиции `dds.order_stage_rep` на кластере, +потом одним запросом пишет весь отрезок через `dds.order_stage_dist`. Она +читает текущее состояние `ods.order_v`, считает `order_date` и переводит +времена в `Europe/Samara`, разбирает `items` в три параллельных массива и +ставит координаты собственного запуска. Distributed-вставка должна +завершиться до замены и разложить строки по ключу цели. + +Замена работает с локальными `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` - покрыть +непрерывный отрезок от него до конца текущего окна. После успешного прогона +число заказов и контрольная сумма выбранного дня должны совпасть с +сохраненными. + +## Как проверить себя + +В `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, включая дни только одной стороны. + +Напишите отдельный запрос к `dds.order_dist`, который получает фактическую +ноду через `hostName()`, вычисляет ожидаемую по `cityHash64(order_id) % 2` и +считает несовпадения по каждой ноде. На обеих нодах результат должен быть +нулевым. Счетные DQ читают Distributed-таблицы и ошибку раскладки не увидят. + +Запустите `etl_pipeline` еще раз при неизменном модельном времени. Группа +заказов снова пересоберет текущее изменяемое окно. Контрольные суммы деловых +полей должны остаться прежними, а все семь проверок - зелеными. + +## Приемка + +- [ ] две новые проверки встроены в `dq_check`, были красными на своих + управляемых ошибках и вернулись в зеленое состояние после восстановления; +- [ ] до и после переписывания ODS срез дал баланс `STG = годные + брак`, три + класса ошибок, верную текущую строку и верный шард; +- [ ] переписаны только `order_parse.sql` и три `sql/dds/order_*.sql`, а + несущая инфраструктура приема и трансформации сохранена; +- [ ] удаленная партиция закрытого дня восстановлена вместе с непрерывным + хвостом, ее число строк и контрольная сумма совпали; +- [ ] `order_items_total`, `orders_vs_ods` и пять прежних DQ зелены; +- [ ] проверки `hostName()` не нашли заказов ODS и DDS на чужом шарде; +- [ ] повторный `etl_pipeline` сохранил деловые данные и зеленые проверки; +- [ ] строки обоих опытов удалены только по своим маркерам. diff --git a/docs/labs/README.md b/docs/labs/README.md index b29124a..18b699d 100644 --- a/docs/labs/README.md +++ b/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) | написать проверки качества до рискованного изменения, а не после первой поломки | | 6 | Разбери сырые события и сверь результат со стендом | изучить разбор потока отдельно от живого приема и сверить его со стендовым результатом | | 7 | Мониторинг: куда смотреть | выбрать, куда и на что смотреть в живой системе | -ТЗ для ступеней 5-7 появятся по мере продвижения программы. +ТЗ для ступеней 6-7 появятся по мере продвижения программы. ## Ступень 1: пять витрин слоя DM -- 2.54.0 From 596bcba01e0675b2bbfccd2b908384f670a9b5f9 Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Sun, 6 Sep 2026 16:25:04 -0400 Subject: [PATCH 2/2] =?UTF-8?q?docs(labs):=20=D1=80=D0=B0=D0=B7=D0=B4?= =?UTF-8?q?=D0=B5=D0=BB=D0=B5=D0=BD=D0=BE=20=D0=B7=D0=B0=D0=B4=D0=B0=D0=BD?= =?UTF-8?q?=D0=B8=D0=B5=20=D0=BF=D0=BE=20=D0=B7=D0=B0=D0=BA=D0=B0=D0=B7?= =?UTF-8?q?=D0=B0=D0=BC=20=D0=BD=D0=B0=20=D1=82=D1=80=D0=B8=20=D1=88=D0=B0?= =?UTF-8?q?=D0=B3=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Зачем: - одно длинное ТЗ заставляло одновременно держать в голове три самостоятельных опыта и их очистку. - Что: - проверки, разбор ODS и сборка DDS разнесены по трем последовательным заданиям. - уточнены границы учебной реализации, место диагностики и выбор закрытого дня. - оглавление ступени 5 дополнено ссылками на все три задания. - Проверка: - проверены ссылки, пробная пачка jq, кластерные мутации и отсутствие ошибок пробелов в diff. --- docs/labs/5-1-order-checks.md | 311 ++++++-------------------------- docs/labs/5-2-order-parsing.md | 181 +++++++++++++++++++ docs/labs/5-3-order-pipeline.md | 124 +++++++++++++ docs/labs/README.md | 2 +- 4 files changed, 362 insertions(+), 256 deletions(-) create mode 100644 docs/labs/5-2-order-parsing.md create mode 100644 docs/labs/5-3-order-pipeline.md diff --git a/docs/labs/5-1-order-checks.md b/docs/labs/5-1-order-checks.md index 2abfd85..e73bcc2 100644 --- a/docs/labs/5-1-order-checks.md +++ b/docs/labs/5-1-order-checks.md @@ -1,53 +1,44 @@ # Сначала сторож, потом ломай: заказы -**Чему учит:** ставить проверку качества до рискованного изменения и -доказывать управляемой ошибкой, что она защищает нужный договор. +**Чему учит:** до рискованного изменения задавать проверке явный договор и +испытывать его управляемой ошибкой. ## Сначала прочитать -- [прием заказов в ODS](../architecture/orders/ingestion.md); - [модель заказа DDS](../architecture/dds/order.md); - [дневной конвейер](../architecture/etl.md#объём-прогона); -- [раскладку по шардам](../architecture/storage.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` и красит именованные задачи. + `dm.dq_summary` и создает для них отдельные задачи. -Только после красного и зеленого опыта с каждым сторожем перепишите готовые -преобразования: +На этом шаге не меняйте разбор ODS и сборку DDS. Даги, DDL, таблицы-доноры, +Kafka и службы остаются несущей частью упражнения. -- `sql/ods/order_parse.sql`; -- `sql/dds/order_scope.sql`; -- `sql/dds/order_rebuild.sql`; -- `sql/dds/order_replace.sql`. - -Даги приема и трансформации, DDL, таблицы-доноры, Kafka и службы остаются -несущей частью упражнения. Их интерфейсы уже заданы. Перед опытами остановите -автоматическое движение мира: поставьте `world_live` на паузу и дождитесь -завершения активных прогонов. До конца очередного опыта не запускайте -`world_next_day`, `world_live_day` и `world_initialize` руками. Это условие -стабильного модельного времени: одной паузы `world_live` для него мало. Даг -`world_recreate` тоже оставьте нетронутым: он пересоздает весь прикладной мир. +Перед опытами остановите автоматическое движение мира: поставьте `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`. Даг записывает результат под отдельным -`check_name` и падает, если `failed_rows` больше нуля. Диагностика называет -день, `order_id` или сравниваемые дневные значения. Ограничьте ее тем же -малым числом примеров, что и существующие запросы. +`failed_rows`, `diagnostics`. Даг сохраняет в `dm.dq_summary` только счетчики, +статус и координаты запуска. Ограниченные примеры из `diagnostics` ищите в +журнале упавшей задачи Airflow. Ограничьте их тем же малым числом, что и +существующие запросы в `sql/dq/`. ### Сумма позиций @@ -55,13 +46,12 @@ посчитайте сумму `item_price * item_quantity` и сравните ее с `items_total`. `reference_rows` считает ключи со стороны `items_total`, `actual_rows` - ключи со стороны вычисленной суммы, а `failed_rows` - несовпавшие значения. В -исправном наборе первые два счетчика равны по построению; сам инвариант стережет -`failed_rows`. В диагностике достаточно `order_id`, ожидаемой суммы и суммы -позиций. +исправном наборе первые два счетчика равны по построению; сам инвариант выражен +через `failed_rows`. Для диагностики достаточно `order_id`, ожидаемой суммы и +суммы позиций. -Эта проверка стережет молчаливый типизированный разбор массива `items` на -границе ODS -> DDS. Она не проверяет скидку, доставку или равенство `total`: -это другие договоры заказа. +Проверка относится к типизированному разбору массива `items` на границе +ODS -> DDS. Скидка, доставка и равенство `total` в этот договор не входят. ### DDS против ODS @@ -72,24 +62,24 @@ - число заказов; - максимальный `updated_at`. -День ODS вычисляется в `Europe/Samara`, как `order_date` в DDS. Постройте -область из объединения дней обеих сторон: полностью пропавший день тоже должен -остаться в ответе. Перед сравнением приведите оба времени к одному часовому +День ODS вычисляется в `Europe/Samara`, как `order_date` в DDS. Область +проверки - объединение дней обеих сторон, включая день, который полностью +пропал из одного слоя. Перед сравнением приведите времена к одному часовому поясу. `reference_rows` - число заказов ODS, `actual_rows` - число заказов DDS, -а `failed_rows` равен единице для дня, где расходится хотя бы одно из двух -значений. Обе стороны распределены по `order_id`, поэтому дневные агрегаты -соедините через `GLOBAL FULL OUTER JOIN` с `join_use_nulls = 1`, как в -существующих сверках. +а `failed_rows` равен единице для дня, где расходится хотя бы одно значение. -Если разбор ODS и последующая сборка DDS одинаково потеряют строку, эта сверка -согласится с обоими. От общей потери при разборе защищает отдельный баланс по -`_load_id`: его вы проверите до запуска нижележащих преобразований. +Обе стороны распределены по `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` у обоих имен не должно быть нарушенных -дней. +завершиться успешно, а в `dm.dq_summary_v` у обоих имен не должно быть +нарушенных дней. Выберите оплаченный заказ из последнего слепка с непустым массивом позиций и сохраните его `order_id` и `order_date`. Подключитесь к первой ноде из DBeaver @@ -104,13 +94,14 @@ WHERE order_id = '' SETTINGS mutations_sync = 2; ``` -Затем запустите только `dq_check`. +Затем запустите только `dq_check`. Задача `order_items_total` должна +завершиться ошибкой, а ее журнал - назвать выбранный заказ. +`purchase_vs_orders_vs_dds` тоже даст красный результат: сохраненная витрина +содержит прежний `items_total`. -`order_items_total` должен стать красным и назвать выбранный заказ. -`purchase_vs_orders_vs_dds` тоже покраснеет: его сохраненная витрина еще несет -прежний `items_total`. Исправьте опыт запуском `etl_pipeline`. Заказы -пересобирают всё текущее окно, и строка вернется из ODS. Убедитесь, что новый -сторож снова зеленый. +Запустите `etl_pipeline`. Текущее окно заказов будет заново собрано из ODS, и +исходная строка вернется в DDS. Повторный `dq_check` должен дать зеленый +результат для нового сторожа. Теперь удалите тот же заказ из `dds.order_rep` на всем кластере: @@ -120,193 +111,11 @@ DELETE WHERE order_id = '' SETTINGS mutations_sync = 2; ``` -До восстановления запустите только `dq_check`: `orders_vs_ods`, -`purchase_vs_orders_vs_dds` и `revenue_daily_vs_dds` должны покраснеть на дне -оплаченного заказа. Затем запустите `etl_pipeline` и получите зеленый -результат. Если новая проверка не увидела свою поломку, чините ее до перехода -к разбору. - -## Проверьте обе ветви разбора - -Чистый мир не производит брак заказов. Поэтому до вычитания -`order_parse.sql` проведите отдельный опыт, а после своей реализации повторите -его с новым маркером. - -Сделайте четыре сообщения с общим уникальным маркером: - -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` проверьте этот срез: - -- в 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-таблицам -докажите, что строк маркера не осталось, и удалите временный файл. Сообщения -в Kafka уже подтверждены группой `clickstream_orders` и повторно не читаются. -В журнале они останутся до удаления по политике хранения топика. -Очистка должна закончиться до любого запуска `etl_pipeline`. Если конвейер все -же успел собрать временный заказ в DDS, после очистки запустите его еще раз: -обычная пересборка текущего окна уберет строку маркера. - -## Перепишите разбор ODS - -Удалите готовый SQL из `sql/ods/order_parse.sql` и соберите переход заново. -Его интерфейс остается прежним: - -- вход - строки `stg.orders_raw_dist` одного `{load_id:String}`; -- первая вставка пишет годные строки в `ods.order_dist`; -- вторая пишет остальные строки в `ods.order_errors_dist`; -- классы брака и их порядок - `not_an_object`, `keyset_mismatch`, - `field_invalid`; -- обе ветви используют один предикат, который на произвольном `raw` дает - только 0 или 1 и не бросает исключение; -- метки доставки и `_load_id` переходят из STG без пересчета; -- обе вставки дожидаются Distributed-записи и отдают раскладку ключу цели. - -Точный набор полей, типы и граница строгого приема описаны в -[документе приема](../architecture/orders/ingestion.md#граница-строгого-приёма), -а целевые колонки - в [`20-ods-tables.sql`](../../sql/ddl/20-ods-tables.sql). -Внутренность `items` на этой границе остается сырым JSON: ее разбирает DDS. - -Повторите опыт четырех строк с новым маркером. Сначала получите тот же баланс, -три класса, текущую версию по `snapshot_date`, разные времена и правильный -шард. Только затем очистите срез и переходите в DDS. - -## Перепишите модель DDS - -Удалите готовый SQL из трех файлов `sql/dds/order_*.sql` и сохраните договор с -`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` равен -началу окна. Если дыра старше, он расширяет непрерывный отрезок влево до нее. - -Пересборка сначала очищает все партиции `dds.order_stage_rep` на кластере, -потом одним запросом пишет весь отрезок через `dds.order_stage_dist`. Она -читает текущее состояние `ods.order_v`, считает `order_date` и переводит -времена в `Europe/Samara`, разбирает `items` в три параллельных массива и -ставит координаты собственного запуска. Distributed-вставка должна -завершиться до замены и разложить строки по ключу цели. - -Замена работает с локальными `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` - покрыть -непрерывный отрезок от него до конца текущего окна. После успешного прогона -число заказов и контрольная сумма выбранного дня должны совпасть с -сохраненными. +До восстановления снова запустите только `dq_check`. Красный результат нужен +для `orders_vs_ods`, `purchase_vs_orders_vs_dds` и `revenue_daily_vs_dds` на +дне оплаченного заказа. Затем запустите `etl_pipeline` и повторите проверку. +Если `orders_vs_ods` не обнаружил удаление, исправьте запрос до следующего +задания. ## Как проверить себя @@ -322,26 +131,18 @@ ORDER BY check_name; Проверка суммы должна охватывать каждый день `dds.order_v`. Сверка слоев должна охватывать объединение дней ODS и DDS, включая дни только одной стороны. -Напишите отдельный запрос к `dds.order_dist`, который получает фактическую -ноду через `hostName()`, вычисляет ожидаемую по `cityHash64(order_id) % 2` и -считает несовпадения по каждой ноде. На обеих нодах результат должен быть -нулевым. Счетные DQ читают Distributed-таблицы и ошибку раскладки не увидят. +## Когда можно остановиться -Запустите `etl_pipeline` еще раз при неизменном модельном времени. Группа -заказов снова пересоберет текущее изменяемое окно. Контрольные суммы деловых -полей должны остаться прежними, а все семь проверок - зелеными. +После двух восстановлений выбранный заказ снова совпадает с ODS. Все семь +проверок дают зеленый результат. В таком состоянии можно закончить работу или +перейти к разбору заказов, оставив модельное время на паузе. ## Приемка -- [ ] две новые проверки встроены в `dq_check`, были красными на своих - управляемых ошибках и вернулись в зеленое состояние после восстановления; -- [ ] до и после переписывания ODS срез дал баланс `STG = годные + брак`, три - класса ошибок, верную текущую строку и верный шард; -- [ ] переписаны только `order_parse.sql` и три `sql/dds/order_*.sql`, а - несущая инфраструктура приема и трансформации сохранена; -- [ ] удаленная партиция закрытого дня восстановлена вместе с непрерывным - хвостом, ее число строк и контрольная сумма совпали; +- [ ] две новые проверки встроены в `dq_check` и публикуют дневные строки по + принятому договору; +- [ ] каждая проверка дала красный результат на своей управляемой ошибке; +- [ ] журналы упавших задач содержали ограниченную диагностику; +- [ ] `etl_pipeline` восстановил обе измененные строки DDS; - [ ] `order_items_total`, `orders_vs_ods` и пять прежних DQ зелены; -- [ ] проверки `hostName()` не нашли заказов ODS и DDS на чужом шарде; -- [ ] повторный `etl_pipeline` сохранил деловые данные и зеленые проверки; -- [ ] строки обоих опытов удалены только по своим маркерам. +- [ ] SQL разбора ODS и сборки DDS на этом шаге не менялся. diff --git a/docs/labs/5-2-order-parsing.md b/docs/labs/5-2-order-parsing.md new file mode 100644 index 0000000..1662017 --- /dev/null +++ b/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 находятся в зеленом состоянии. diff --git a/docs/labs/5-3-order-pipeline.md b/docs/labs/5-3-order-pipeline.md new file mode 100644 index 0000000..cd5b867 --- /dev/null +++ b/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` сохранил деловые данные и зеленые проверки. diff --git a/docs/labs/README.md b/docs/labs/README.md index 18b699d..4aaaf29 100644 --- a/docs/labs/README.md +++ b/docs/labs/README.md @@ -12,7 +12,7 @@ | 2 | [Свой график в Superset](2-1-superset-chart.md) | превратить свой результат из хранилища в отчет в BI | | 3 | [Нарежь сессии сам](3-1-sessions-by-hand.md) | сверить число сессий в своей нарезке с эталоном и найти события, из-за которых оно расходится | | 4 | [Перепиши нарезку стенда](4-1-session-pipeline.md) | изменить расчетную сущность внутри готового конвейера и проверить ее внешним эталоном | -| 5 | [Сначала сторож, потом ломай: заказы](5-1-order-checks.md) | написать проверки качества до рискованного изменения, а не после первой поломки | +| 5 | [Сначала сторож, потом ломай: заказы](5-1-order-checks.md)
[Перепишите разбор заказов в ODS](5-2-order-parsing.md)
[Перепишите сборку заказов в DDS](5-3-order-pipeline.md) | написать проверки качества до рискованного изменения, затем переписать разбор и сборку заказов | | 6 | Разбери сырые события и сверь результат со стендом | изучить разбор потока отдельно от живого приема и сверить его со стендовым результатом | | 7 | Мониторинг: куда смотреть | выбрать, куда и на что смотреть в живой системе | -- 2.54.0