clickstream-data-platform/docs/labs/5-1-order-checks.md

Сначала сторож, потом ломай: заказы

Чему учит: ставить проверку качества до рискованного изменения и доказывать управляемой ошибкой, что она защищает нужный договор.

Сначала прочитать

• прием заказов в ODS;
• модель заказа DDS;
• дневной конвейер;
• раскладку по шардам;
• правила проверок.

Посмотрите также на устройство одной проверки в sql/dq/ и на ее включение в dq_check. Готового ответа для проверок из этого задания в репозитории нет.

Граница задания

Сначала добавьте два сторожа:

• 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 под пользователем etl. Прибавьте одну копейку к items_total этого заказа в локальной таблице dds.order_rep на всем кластере:

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.order_rep на всем кластере:

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 и получите зеленый результат. Если новая проверка не увидела свою поломку, чините ее до перехода к разбору.

Проверьте обе ветви разбора

Чистый мир не производит брак заказов. Поэтому до вычитания 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. Следующая заготовка дает одну годную строку и по одному примеру трех классов; уникальный маркер не смешивает опыт с данными мира:

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"

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

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.

После записи результатов удалите только строки маркера из локальных таблиц:

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-записи и отдают раскладку ключу цели.

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

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 должны быть зелены обе новые и пять прежних проверок:

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 сохранил деловые данные и зеленые проверки;
• строки обоих опытов удалены только по своим маркерам.