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

Проверки качества заказов

Перед изменением обработки заказов вы добавите две проверки качества данных. Затем намеренно испортите один заказ и убедитесь, что проверки обнаружили ошибки.

Подготовка

• модель заказа 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 и создает для них отдельные задачи.

Разбор ODS и сборку DDS вы перепишете в следующих заданиях. Здесь сохраните их, даги загрузки, DDL, промежуточные таблицы, Kafka и службы как есть.

Перед опытами остановите автоматическое движение мира: поставьте world_live на паузу и дождитесь завершения активных прогонов. До завершения опытов не запускайте руками world_next_day, world_live_day, world_initialize и world_recreate, чтобы новые данные не меняли результат между проверками.

Требования к запросам

Чтобы 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_total и вычисленной суммы позиций.

Проверка поможет заметить ошибки разбора items при переносе заказа из ODS в DDS. Скидку, доставку и равенство total здесь проверять не нужно.

Полнота заказов в DDS

Имя проверки - orders_vs_ods. Источник модели DDS здесь - текущее состояние ods.order_v. С ним вы и сравниваете DDS; обращаться к внутреннему состоянию генератора для этой проверки не нужно. Одним запросом рассчитайте итоги каждого дня заказа в обоих слоях и сравните:

• число заказов;
• максимальный updated_at.

День заказа в ODS вычисляйте в Europe/Samara, как order_date в DDS. Проверьте все дни, которые есть хотя бы в одном слое: так вы заметите день, целиком пропавший из другого. Перед сравнением приведите времена к одному часовому поясу. reference_rows содержит число заказов ODS, actual_rows число заказов DDS. В failed_rows запишите 1, если за день расходится хотя бы одно значение, и 0, если оба совпали.

Обе стороны распределены по 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. Повторный dq_check должен дать зеленый результат для новой проверки.

Теперь удалите тот же заказ из 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 и повторите проверку. Если orders_vs_ods не обнаружил удаление, исправьте запрос до следующего задания.

Проверка результата

В 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, включая дни только одной стороны.

Завершение работы

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

Результат работы

• две новые проверки запускаются в dq_check, а их дневные результаты появляются в dm.dq_summary;
• вы увидели ошибку каждой проверки после соответствующего изменения заказа, а в журнале нашли примеры заказов с расхождениями;
• вы восстановили заказ через etl_pipeline после обоих опытов;
• order_items_total, orders_vs_ods и пять прежних проверок успешны;
• разбор ODS и сборка DDS остались прежними.