docs(labs): разделено задание по заказам на три шага
- Зачем: - одно длинное ТЗ заставляло одновременно держать в голове три самостоятельных опыта и их очистку. - Что: - проверки, разбор ODS и сборка DDS разнесены по трем последовательным заданиям. - уточнены границы учебной реализации, место диагностики и выбор закрытого дня. - оглавление ступени 5 дополнено ссылками на все три задания. - Проверка: - проверены ссылки, пробная пачка jq, кластерные мутации и отсутствие ошибок пробелов в diff.
This commit is contained in:
ddadmin committed
1 parent
d686de516b
commit
596bcba01e
4 files changed
+362
-256
No files matched your search
+56
-255
@@ -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 = '<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 = '<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 на этом шаге не менялся.
|
||||
@@ -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 находятся в зеленом состоянии.
|
||||
@@ -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` сохранил деловые данные и зеленые проверки.
|
||||
+1
-1
@@ -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)<br>[Перепишите разбор заказов в ODS](5-2-order-parsing.md)<br>[Перепишите сборку заказов в DDS](5-3-order-pipeline.md) | написать проверки качества до рискованного изменения, затем переписать разбор и сборку заказов |
|
||||
| 6 | Разбери сырые события и сверь результат со стендом | изучить разбор потока отдельно от живого приема и сверить его со стендовым результатом |
|
||||
| 7 | Мониторинг: куда смотреть | выбрать, куда и на что смотреть в живой системе |
|
||||
|
||||
|
||||
Reference in new issue
Block a user