diff --git a/Makefile b/Makefile index 2f5b8a4..f340831 100644 --- a/Makefile +++ b/Makefile @@ -13,6 +13,7 @@ GENERATOR_SPEED ?= up: $(COMPOSE) up --detach --build --wait --wait-timeout 600 COMPOSE_BIN="$(COMPOSE)" ./scripts/wait-for-world.sh + COMPOSE_BIN="$(COMPOSE)" ./scripts/ingest-starting-orders.sh down: $(COMPOSE) down --remove-orphans diff --git a/README.md b/README.md index 9bf157d..5df513a 100644 --- a/README.md +++ b/README.md @@ -126,13 +126,20 @@ Kafka по той же причине спрашивают снаружи. Её Стенд поднимается не пустым: разовая служба `world-init` играет в топик `hits` первые восемь дней модельного времени — понедельник по понедельник, 401 185 -событий. Дальше их обычным путём разбирает хранилище, и к концу `make up` они -лежат в `ods.event`. Так у всякой лабы есть данные, и всегда одни и те же. +событий. Следом второй запуск того же генератора отправляет в топик `orders` +семь слепков — за дни 0…6. Дальше оба источника разбирает хранилище, и к концу +`make up` события лежат в `ods.event`, а заказы — в `ods.order_v`. Так у всякой +лабы есть данные, и всегда одни и те же. -Ждать приходится дольше, чем работает заливка: приём асинхронный, поэтому -вторым шагом `make up` зовёт `scripts/wait-for-world.sh` — тот опрашивает -ClickHouse, пока мир не доедет. Повторный `make up` заливает мир заново; это -не ошибка, а свойство: номера событий те же, и повтор схлопнет +Последний день намеренно неполон: события дня 7 уже приехали, а его слепок +уедет только с первым ходом мира. Поэтому на свежем стенде правый край графика +выручки отстаёт от кликстрима и дозаполняется по мере движения мира — так +виден разный темп потокового трекера и ночной выгрузки магазина. + +Ждать приходится дольше, чем работает заливка. `make up` сначала опрашивает +ClickHouse, пока события не доедут, затем разово запускает даг `orders_ingest` +и ждёт его конца. Повторный `make up` заливает мир заново; это не ошибка, а +свойство: номера событий и версии заказов те же, и повторы схлопнут `ReplacingMergeTree`. Сам мир в git не хранится — он чистая функция зерна, и держать его в diff --git a/compose.yaml b/compose.yaml index a8b2218..d8a208f 100644 --- a/compose.yaml +++ b/compose.yaml @@ -36,11 +36,19 @@ x-clickhouse-common: &clickhouse-common # пульт мира. Названы по одному разу — иначе однажды разойдутся, и заметит это # не проверка, а менти с пустым топиком. x-generator-image: &generator-image clickstream-generator:local +x-orders-topic: &orders-topic orders + +x-kafka-common: &kafka-common + KAFKA_BOOTSTRAP_SERVERS: kafka:9092 x-kafka-target: &kafka-target - KAFKA_BOOTSTRAP_SERVERS: kafka:9092 + <<: *kafka-common KAFKA_TOPIC: hits +x-orders-kafka-target: &orders-kafka-target + <<: *kafka-common + KAFKA_TOPIC: *orders-topic + x-world-starting-days: &world-starting-days "8" x-airflow-common: &airflow-common @@ -71,7 +79,7 @@ x-airflow-common: &airflow-common GENERATOR_IMAGE: *generator-image STAND_NETWORK: ${COMPOSE_PROJECT_NAME}_default WORLD_STARTING_DAYS: *world-starting-days - KAFKA_ORDERS_TOPIC: orders + KAFKA_ORDERS_TOPIC: *orders-topic volumes: - ./dags:/opt/airflow/dags:ro # Запросы дагов: даг называет файл, а текст читает Airflow при исполнении @@ -307,6 +315,17 @@ services: <<: *generator-common command: ["batch", "--day", "0", "--days", *world-starting-days] + # Второй источник стартового мира идёт после событий тем же генератором. + # Сдвиг команды snapshot оставляет день 7 первому ходу мира: за восемь + # дней уезжают семь слепков, дней 0…6. + world-snapshots: + <<: *generator-common + depends_on: + world-init: + condition: service_completed_successfully + environment: *orders-kafka-target + command: ["snapshot", "--day", "0", "--days", *world-starting-days] + # Тот же образ для ручных прогонов: `make generate-batch`, `make generate-live`. # Под профилем — чтобы обычный подъём стенда её не трогал. generator: @@ -355,10 +374,9 @@ services: # упавшим для --wait. clickhouse-init: condition: service_completed_successfully - # По той же причине. Заливка стартового мира идёт последней в цепи - # разовых служб, и зависимого ей взять больше негде: долгоживущие службы - # данных не ждут, а `airflow-init` эту цепь и так замыкает. - world-init: + # По той же причине. Слепки идут за событиями и замыкают цепь разовых + # служб; долгоживущие службы данных её не ждут, а `airflow-init` ждёт. + world-snapshots: condition: service_completed_successfully entrypoint: ["/bin/bash"] command: ["/opt/airflow/init.sh"] diff --git a/dags/orders_ingest.py b/dags/orders_ingest.py index e655077..566c68b 100644 --- a/dags/orders_ingest.py +++ b/dags/orders_ingest.py @@ -101,8 +101,8 @@ def orders_ingest(): logging.info("срез разобран: _load_id %s", load_id) pull_batch("stg/orders_raw_load.sql") >> parse_batch( - "ods/order_snapshot_load.sql", - "ods/order_snapshot_errors_load.sql", + "ods/order_load.sql", + "ods/order_errors_load.sql", ) diff --git a/docs/adr/0010-order-versions-in-ods.md b/docs/adr/0010-order-versions-in-ods.md index 86d6b70..8d5fca2 100644 --- a/docs/adr/0010-order-versions-in-ods.md +++ b/docs/adr/0010-order-versions-in-ods.md @@ -9,7 +9,7 @@ команде Airflow, но порция чтения больше не считается полным слепком и не публикуется заменой партиции `snapshot_date`. Прочитанные строки получают `_load_id` запуска, разбираются из одного среза STG в годные строки и ошибки, -а `ods.order_snapshot` хранит принятые версии заказа в +а `ods.order` хранит принятые версии заказа в `ReplacingMergeTree(updated_at)`. Ключ сущности — `order_id`, партиция строится от неизменного `created_at`, все версии ключа направляются на один шард. diff --git a/docs/architecture/orders/README.md b/docs/architecture/orders/README.md index 3848079..1f7c9ef 100644 --- a/docs/architecture/orders/README.md +++ b/docs/architecture/orders/README.md @@ -37,7 +37,7 @@ `person_id` заказ показывает как `user_id`, кликстрим остаётся анонимным — [мост к склейке](identity.md). - **Приём — пакетный забор**: одно прямое чтение Kafka в STG, два - `INSERT SELECT` в типизированный ODS и таблицу ошибок; `ods.order_snapshot` + `INSERT SELECT` в типизированный ODS и таблицу ошибок; `ods.order` принимает версии заказа на `ReplacingMergeTree(updated_at)` — [приём из Kafka в ODS](ingestion.md). - **Стартовый мир отправляет семь слепков** (дни 0…6): у последнего прожитого diff --git a/docs/architecture/orders/fate.md b/docs/architecture/orders/fate.md index 4a99c94..117aee7 100644 --- a/docs/architecture/orders/fate.md +++ b/docs/architecture/orders/fate.md @@ -61,7 +61,7 @@ **Что из двух дорог видно.** В `dds.order` дороги неразличимы — там последняя версия; различает их история версий: сырьё STG и физические версии -`ods.order_snapshot` до фоновых слияний, а в витринах — выручка дня, которая +`ods.order` до фоновых слияний, а в витринах — выручка дня, которая сначала выросла, потом убыла. Полное различение не обещано: слепок — состояние на границе суток, и оплата с отменой в один день в сырье неразличимы; то же у сильно опоздавших, приехавших уже терминальными. Точную форму этого урока diff --git a/docs/architecture/orders/ingestion.md b/docs/architecture/orders/ingestion.md index c628339..1218932 100644 --- a/docs/architecture/orders/ingestion.md +++ b/docs/architecture/orders/ingestion.md @@ -43,12 +43,12 @@ Один следующий `task_id` отвечает за весь переход STG → ODS. Внутри него два последовательных `INSERT SELECT` читают неизменный срез по `_load_id`: первый -пишет годные строки в `ods.order_snapshot`, второй — брак в -`ods.order_snapshot_errors`. Транзакции между запросами нет. При частичном сбое -Airflow повторяет весь `task_id`; одинаковые исходные строки и служебные метки -не вычисляются заново. Запросы лежат рядом с целями: -[`order_snapshot_load.sql`](../../../sql/ods/order_snapshot_load.sql) и -[`order_snapshot_errors_load.sql`](../../../sql/ods/order_snapshot_errors_load.sql). +пишет годные строки в `ods.order`, второй — брак в `ods.order_errors`. +Транзакции между запросами нет. При частичном сбое Airflow повторяет весь +`task_id`; одинаковые исходные строки и служебные метки не вычисляются заново. +Запросы лежат рядом с целями: +[`order_load.sql`](../../../sql/ods/order_load.sql) и +[`order_errors_load.sql`](../../../sql/ods/order_errors_load.sql). Условия запросов взаимодополняющие: один общий предикат определяет брак, а годная ветвь использует его буквальное отрицание. Все функции предиката @@ -90,9 +90,9 @@ JSON-объектом с точным набором ключей: `order_id`, ` ## Роль ODS -`ods.order_snapshot_rep` хранит физически принятые версии в +`ods.order_rep` хранит физически принятые версии в `ReplacingMergeTree(updated_at)`. Ключ сортировки — `order_id`, партиция — день -неизменного `created_at`. `ods.order_snapshot_dist` шардирует по +неизменного `created_at`. `ods.order_dist` шардирует по `cityHash64(order_id)`: только так все версии заказа попадают на один шард и `FINAL` даёт корректный результат через распределённую таблицу. @@ -174,7 +174,7 @@ JSON-объектом с точным набором ключей: `order_id`, ` После исправления [#102](https://git.dementev.space/ddmitry/clickstream-data-platform/issues/102) честный прогон дня снят на живом стенде 20 августа 2026 года: все 1694 строки -слепка дня 7 приняты, `ods.order_snapshot_errors` осталась пустой. У заказа +слепка дня 7 приняты, `ods.order_errors` осталась пустой. У заказа `20260608-0007`, который прежде приходил с отрицательным итогом, теперь `5490.00 − 823.50 + 299.00 = 4965.50`. diff --git a/docs/architecture/orders/inventory.md b/docs/architecture/orders/inventory.md index e790ee5..5ddba60 100644 --- a/docs/architecture/orders/inventory.md +++ b/docs/architecture/orders/inventory.md @@ -25,6 +25,6 @@ пользователей и число двухкуковых пар растут, пока мир едет, — опоры из них не выходит; генератор сторожит хеш, транспорт — счёт событий. - **Опись описывает мир, а не доставку.** Работа генератора кончается на - Kafka: дошли ли байты до `ods.order_snapshot` — вопрос стенда и его + Kafka: дошли ли байты до `ods.order` — вопрос стенда и его проверок. Поэтому счёта строк у слепка в описи нет; понадобится проверка приёма заказов — число заведётся вместе с ней. diff --git a/docs/architecture/storage.md b/docs/architecture/storage.md index 2dad3a2..5702a25 100644 --- a/docs/architecture/storage.md +++ b/docs/architecture/storage.md @@ -60,7 +60,7 @@ CSV-файла. Живёт он вне цепочки слоёв, в зоне `d приходит от транспорта: слой хранит то, что доехало по топику, и зовётся именем топика — `hits_raw` от `hits`, `orders_raw` от `orders`. В типизированных слоях имя приходит от предметной области и стоит в единственном числе: `event`, -`order_snapshot`, `session`. Граница между «как привезли» и «что это такое» +`order`, `session`. Граница между «как привезли» и «что это такое» проходит по STG, и имена её показывают. ## Раскладка по шардам @@ -91,7 +91,7 @@ CSV-файла. Живёт он вне цепочки слоёв, в зоне `d по `cityHash64(order_id)`; ключи остальных частей будущей модели и витрин появятся вместе с ними. -Замена партиций не используется для `ods.order_snapshot`: версии прошлых дней +Замена партиций не используется для `ods.order`: версии прошлых дней доливаются, а прямое чтение Kafka не задаёт границы полного слепка ([ADR 0010](../adr/0010-order-versions-in-ods.md)). Если партиционная пересборка понадобится будущим объектам DDS или DM, партиция-донор должна быть заранее @@ -141,7 +141,7 @@ kafka_offset)`: разбор полётов идёт от «какое сооб заказов — пакетным шагом. Дальше метка переносится в ODS как есть и отвечает на вопрос «когда строка приехала в хранилище», а не «когда её разобрали». В `ods.event` она же служит колонкой версии `ReplacingMergeTree` и схлопывает -повтор доставки. У `ods.order_snapshot` версию задаёт `updated_at` источника; +повтор доставки. У `ods.order` версию задаёт `updated_at` источника; `_load_ts` только показывает, когда конкретная строка приехала. `created_at` и `updated_at` заказа к служебным колонкам хранилища не относятся. @@ -181,9 +181,9 @@ snake_case-имена и расшифровку кодов. ## Версии заказов -`ods.order_snapshot_rep` хранит принятые версии в +`ods.order_rep` хранит принятые версии в `ReplacingMergeTree(updated_at)`: ключ сортировки — `order_id`, партиция — -`toDate(created_at)`. `ods.order_snapshot_dist` шардирует по +`toDate(created_at)`. `ods.order_dist` шардирует по `cityHash64(order_id)`. Все версии заказа лежат в одной партиции, чтобы их могли схлопывать фоновые слияния, и на одном шарде, чтобы распределённый `FINAL` выбрал одного победителя. @@ -414,7 +414,7 @@ D0 и к реальному календарю не привязана; паке kafka_offset)`: смотрят такую таблицу от класса, а внутри класса — по координатам доставки. -`ods.order_snapshot_errors` держит тот же диагностический минимум и `_load_id` +`ods.order_errors` держит тот же диагностический минимум и `_load_id` запуска. У заказов три класса по приоритету: `not_an_object`, `keyset_mismatch`, `field_invalid`. Сырой текст остаётся рядом, поэтому класс не разрастается до имени отдельного поля. Точная граница приёма — в @@ -534,15 +534,15 @@ Airflow читает и собирает эти файлы штатным шаб | `ods` | `ods.event_v` | актуальная версия события с полями источника | | `ods` | `ods.event_errors_rep` / `_dist` | строки, не прошедшие строгий приём | | `ods` | `ods.event_mv`, `ods.event_errors_mv` | разбор сырья в событие и в ошибки | -| `ods` | `ods.order_snapshot_rep` / `_dist` | типизированные версии заказа | +| `ods` | `ods.order_rep` / `_dist` | типизированные версии заказа | | `ods` | `ods.order_v` | текущая версия заказа на языке источника | -| `ods` | `ods.order_snapshot_errors_rep` / `_dist` | строки слепка, не прошедшие строгий приём | +| `ods` | `ods.order_errors_rep` / `_dist` | строки слепка, не прошедшие строгий приём | | `dic` | `dic.products_file` | чтец CSV-каталога, общего с генератором | | `dic` | `dic.products` | словарь товаров поверх подложки | Матвью разбора у заказов нет: срез сырья раскладывают по этим двум целям два -`INSERT SELECT` шага `parse_batch` из файлов `sql/ods/order_snapshot_load.sql` -и `sql/ods/order_snapshot_errors_load.sql`. +`INSERT SELECT` шага `parse_batch` из файлов `sql/ods/order_load.sql` и +`sql/ods/order_errors_load.sql`. Слой DDS и слой DM появляются на следующих этапах; их состав задан разделом 7 мастер-спеки и переносится сюда по мере постройки. @@ -606,14 +606,13 @@ DDL-словарь с файловым источником внутри `user_f `_load_id`, пересечение ветвей — ноль. Все восемь строк брака — версии одного заказа с отрицательным `total`; это дефект генератора, разобранный в [спецификации приёма](orders/ingestion.md), «Что проверено». На управляемой -порции из пяти строк (две версии -одного заказа плюс по одной строке каждого класса брака) годные ушли в -`ods.order_snapshot`, брак — в `ods.order_snapshot_errors` с ожидаемыми -классами. Обе версии легли в одну партицию и на один шард; `FINAL` через -распределённую таблицу вернул одну строку — ту, у которой `updated_at` позже, — -тогда как физический счёт показывал две. Повтор задачи с тем же `_load_id` -строку в `ods.order_v` и её `_load_ts` не изменил, а таблица ошибок записала -тот же брак второй раз, как и обещано спецификацией приёма. +порции из пяти строк (две версии одного заказа плюс по одной строке каждого +класса брака) годные ушли в `ods.order`, брак — в `ods.order_errors` с +ожидаемыми классами. Обе версии легли в одну партицию и на один шард; `FINAL` +через распределённую таблицу вернул одну строку — ту, у которой `updated_at` +позже, — тогда как физический счёт показывал две. Повтор задачи с тем же +`_load_id` строку в `ods.order_v` и её `_load_ts` не изменил, а таблица ошибок +записала тот же брак второй раз, как и обещано спецификацией приёма. Отдельно измерено, что **сводка вставки не годится в счётчики строк**: у запроса с `WHERE` `read_rows` считает прочитанное с диска, а не подошедшее, — diff --git a/docs/architecture/testing.md b/docs/architecture/testing.md index 65ad23f..1145354 100644 --- a/docs/architecture/testing.md +++ b/docs/architecture/testing.md @@ -71,9 +71,9 @@ тогда окружение окупится, и условие возврата — именно это, а не «стало неудобно». -Цена самого `make up` — 2 м 50 с с нуля (`make clean` перед ним — ещё 23 с) и -1 м 5 с на живом стенде. В цену целей она не входит, но записана здесь по той -же причине: с #42 подъём стенда заливает стартовый мир и стал заметно дороже. +Цена самого `make up` — 8 м 18 с с нуля и 3 м 35 с на живом стенде. В цену +целей она не входит, но записана здесь по той же причине: подъём пересчитывает +оба источника стартового мира и ждёт их приёма. ## Правило: смоук обязан оставаться быстрым @@ -194,12 +194,12 @@ Kafka → STG → ODS, и живёт он в `make check-clickhouse`. | `make check-services` | `scripts/stand-services.sh` | | `make check-clickhouse` | `scripts/check-clickhouse.sh` | -Особняком — `scripts/wait-for-world.sh`: он не проверка, а вторая половина -`make up`. `docker compose up --wait` дожидается служб, в том числе успешно -отработавшей заливки, но «заливка кончилась» значит «события в топике», а не -«события в хранилище»: приём асинхронный. Скрипт ждёт доезда ограниченным -циклом опроса. Без него `make check-clickhouse` следом краснел бы по -устройству, а не по поломке. +Особняком — два сценария продолжения `make up`, оба не проверки. +`scripts/wait-for-world.sh` ждёт, пока события доедут из Kafka в ODS: +`docker compose up --wait` видит конец отправки, но приём идёт асинхронно. +Затем `scripts/ingest-starting-orders.sh` разово запускает единственный даг +приёма заказов `orders_ingest` и ждёт его конца. Оба ждут ограниченным циклом, +а не паузой наугад. Общее у смоука и `check-services` — счёт проверок, обращение к Compose и две проверки — вынесено в `scripts/stand-common.sh`; сам он не запускается. @@ -221,6 +221,15 @@ Kafka → STG → ODS, и живёт он в `make check-clickhouse`. ## Что проверено +**Перезамер 20 августа 2026 года при исполнении #96.** `make up` после +`make clean` прошёл за 8 м 18 с, повтор на живом стенде — за 3 м 35 с. Служба +семи слепков заняла около 28 секунд между концом `world-init` и своим концом. +Прежний замер ниже — 2 м 50 с и 1 м 5 с до появления приёма заказов и пульта +мира. Весь рост приписывать слепкам нельзя: на холодном прогоне основное время +ушло ещё до сценариев ожидания: `airflow-init` завершился примерно на 320-й +секунде, обработчик DAG стал здоров примерно на 458-й. Числа здесь называют +цену нынешнего стенда, а не выносят вердикт о регрессии. + **Перезамер 19 августа 2026 года при исполнении #95.** После добавления десятой проверки словаря `make check-clickhouse` на чисто поднятом стенде прошёл за 24 секунды. Это цена всей цели, а не одной новой строки. diff --git a/docs/research/2026-08-16-order-snapshot-wire-format.md b/docs/research/2026-08-16-order-snapshot-wire-format.md index ce2db32..33fa3bf 100644 --- a/docs/research/2026-08-16-order-snapshot-wire-format.md +++ b/docs/research/2026-08-16-order-snapshot-wire-format.md @@ -64,7 +64,7 @@ Debezium проводит ту же границу внутри одного с `created_at` и `updated_at` — аудит строки источника. Поэтому `toDate(created_at)` в [спецификации приёма заказов](../architecture/orders/ingestion.md) -используется как стабильный технический ключ партиции `ods.order_snapshot`: +используется как стабильный технический ключ партиции `ods.order`: это день создания строки источника, а не доказательство дня бизнес-события. diff --git a/docs/specs/2026-07-30-stand-v2-realism.md b/docs/specs/2026-07-30-stand-v2-realism.md index 77e715d..d99f5d8 100644 --- a/docs/specs/2026-07-30-stand-v2-realism.md +++ b/docs/specs/2026-07-30-stand-v2-realism.md @@ -219,7 +219,7 @@ Ecommerce (заполнены только у торговых событий): Обоснование и проверка разбора — в [исследовании формата](../research/2026-08-16-order-snapshot-wire-format.md). -- `ods.order_snapshot` принимает версии заказа в +- `ods.order` принимает версии заказа в `ReplacingMergeTree(updated_at)`: `ORDER BY order_id`, партиция по дню неизменного `created_at`, шардирование по `cityHash64(order_id)`. `snapshot_date` остаётся датой наблюдения источника, но не задаёт публикацию @@ -410,7 +410,7 @@ README. | STG | `stg.hits_raw_kafka`, `stg.hits_raw` + MV | сырые строки событий, Kafka Engine на обеих нодах | | STG | `stg.orders_raw_kafka`, `stg.orders_raw`, без MV | сырые строки слепка; чтец на ноде 1, забирает пакетный шаг | | ODS | `ods.event` (+`_errors`), `ods.event_v` | типизированные версии событий, брак и точное актуальное состояние | -| ODS | `ods.order_snapshot` (+`_errors`), `ods.order_v` | типизированные версии заказов, брак и точное текущее состояние | +| ODS | `ods.order` (+`_errors`), `ods.order_v` | типизированные версии заказов, брак и точное текущее состояние | | DDS | `dds.session` | сборка сессий из событий (наследник `dds.click`) | | DDS | `dds.event_v` | представление над `ods.event`: snake_case-имена, расшифровка кодов `DeviceCategory`; витрины DM читают его, а не ODS напрямую | | DDS | модель заказов | зерно, связи и материализация проектируются на этапе DDS | @@ -428,7 +428,7 @@ README. матвью к нему не привязана, и сырьё забирает шаг, которым управляет Airflow — одно прямое чтение вставляет порцию в `stg.orders_raw` с `_load_id` запуска. Один следующий `task_id` двумя последовательными запросами пишет годные строки -и ошибки из того же среза. `ods.order_snapshot` хранит версии по `updated_at`, +и ошибки из того же среза. `ods.order` хранит версии по `updated_at`, а не публикует партицию `snapshot_date`. Слой сырья остаётся точкой повтора и разбора одной принятой порции. @@ -436,7 +436,7 @@ README. из опорных точек раздела 12 переформулировано под них. Состав служебных колонок задаёт дока хранилища. Спеке важны два следствия: -`ods.event` и `ods.order_snapshot` получают метку загрузки `_load_ts`, но у +`ods.event` и `ods.order` получают метку загрузки `_load_ts`, но у заказа бизнес-версию задаёт `updated_at`; `_load_id` проходит через STG и обе цели ODS. Таблицы `stg.*_raw` хранят метаданные доставки Kafka вместе с именем читавшей ноды — diff --git a/scripts/ingest-starting-orders.sh b/scripts/ingest-starting-orders.sh new file mode 100755 index 0000000..b816914 --- /dev/null +++ b/scripts/ingest-starting-orders.sh @@ -0,0 +1,94 @@ +#!/usr/bin/env bash +# Последний шаг `make up`: запускает единственный путь приёма заказов и ждёт +# его конца. Слепки к этому моменту уже лежат в Kafka, а события доехали до ODS. +set -Eeuo pipefail + +readonly ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +readonly DAG_ID='orders_ingest' +# Потолок каждого ожидания — 300 с: больше чем всемеро от замеренных 40 с +# между запуском сценария и концом дага (#96). Это предел, а не бюджет времени. +readonly ATTEMPTS=100 +readonly PAUSE_SECONDS=3 +readonly REPORT_EVERY=5 + +read -r -a COMPOSE_CMD <<<"${COMPOSE_BIN:-docker compose}" + +fail() { + printf 'ОШИБКА: %s\n' "$1" >&2 + exit 1 +} + +compose() { + "${COMPOSE_CMD[@]}" --project-directory "$ROOT_DIR" "$@" +} + +config="$(compose config --format json)" +user="$(jq -r '.services["airflow-apiserver"].environment.AIRFLOW_ADMIN_USER // empty' <<<"$config")" +password="$(jq -r '.services["airflow-apiserver"].environment.AIRFLOW_ADMIN_PASSWORD // empty' <<<"$config")" +binding="$(compose port airflow-apiserver 8080)" +port="${binding##*:}" +[[ -n "$port" ]] || fail 'не найден отображённый порт API Airflow' + +response="$(curl -sf --max-time 10 -X POST \ + -H 'Content-Type: application/json' \ + -d "$(jq -cn --arg username "$user" --arg password "$password" \ + '{username: $username, password: $password}')" \ + "http://127.0.0.1:${port}/auth/token" 2>/dev/null || true)" +token="$(jq -r '.access_token // empty' <<<"$response")" +[[ -n "$token" ]] || fail 'Airflow не принял учётные данные администратора' + +# После `up --wait` обработчик DAG здоров, но файл мог ещё не попасть в список. +# Ограниченный опрос закрывает эту гонку без паузы наугад. +dag='' +for ((attempt = 1; attempt <= ATTEMPTS; attempt++)); do + dag="$(curl -sf --max-time 10 \ + -H "Authorization: Bearer ${token}" \ + "http://127.0.0.1:${port}/api/v2/dags/${DAG_ID}" \ + 2>/dev/null || true)" + jq -e --arg dag_id "$DAG_ID" '.dag_id == $dag_id' \ + >/dev/null 2>&1 <<<"$dag" && break + sleep "$PAUSE_SECONDS" +done +jq -e --arg dag_id "$DAG_ID" '.dag_id == $dag_id' \ + >/dev/null 2>&1 <<<"$dag" || \ + fail "даг ${DAG_ID} не появился в Airflow за $((ATTEMPTS * PAUSE_SECONDS)) с" + +# В API v2 Airflow 3.3 logical_date=null означает событийный ручной запуск. +# Это уместно для дага без расписания: календарь Airflow не подменяет дату +# слепка, которая уже записана в сообщениях. +response="$(curl -sf --max-time 10 -X POST \ + -H "Authorization: Bearer ${token}" \ + -H 'Content-Type: application/json' \ + -d '{"logical_date":null}' \ + "http://127.0.0.1:${port}/api/v2/dags/${DAG_ID}/dagRuns" \ + 2>/dev/null || true)" +run_id="$(jq -r '.dag_run_id // empty' <<<"$response")" +[[ -n "$run_id" ]] || fail "Airflow не запустил даг ${DAG_ID}" + +encoded_run_id="$(jq -rn --arg value "$run_id" '$value | @uri')" +printf 'Ждём приём стартовых слепков дагом %s...\n' "$DAG_ID" +state='' +for ((attempt = 1; attempt <= ATTEMPTS; attempt++)); do + response="$(curl -sf --max-time 10 \ + -H "Authorization: Bearer ${token}" \ + "http://127.0.0.1:${port}/api/v2/dags/${DAG_ID}/dagRuns/${encoded_run_id}" \ + 2>/dev/null || true)" + state="$(jq -r '.state // empty' <<<"$response")" + case "$state" in + success) + printf 'ЗЕЛЁНО: стартовые слепки приняты в ODS.\n' + exit 0 + ;; + failed) + fail "даг ${DAG_ID} завершился с ошибкой; запуск ${run_id}" + ;; + esac + if ((attempt % REPORT_EVERY == 0)); then + printf 'Приём ещё не завершён: состояние %s, ждём дальше (%d с)...\n' \ + "${state:-неизвестно}" "$((attempt * PAUSE_SECONDS))" + fi + sleep "$PAUSE_SECONDS" +done + +fail "даг ${DAG_ID} не завершился за $((ATTEMPTS * PAUSE_SECONDS)) с: + состояние ${state:-неизвестно}, запуск ${run_id}" diff --git a/sql/ddl/20-ods-tables.sql b/sql/ddl/20-ods-tables.sql index 825c4c9..3e56a52 100644 --- a/sql/ddl/20-ods-tables.sql +++ b/sql/ddl/20-ods-tables.sql @@ -175,7 +175,7 @@ ENGINE = Distributed('clickstream_cluster', 'ods', 'event_errors_rep', cityHash6 -- -- items остаётся сырым фрагментом JSON: приём проверяет только, что это -- массив. Что внутри позиций — забота DDS, а не границы провода. -CREATE TABLE IF NOT EXISTS ods.order_snapshot_rep ON CLUSTER clickstream_cluster +CREATE TABLE IF NOT EXISTS ods.order_rep ON CLUSTER clickstream_cluster ( order_id String, user_id UInt64, @@ -199,9 +199,9 @@ ORDER BY order_id; -- и выбор здесь не про перекос, а про корректность: только так все версии -- одного заказа попадают на один шард, и FINAL через распределённую таблицу -- выбирает одного победителя, а не по победителю на шард. -CREATE TABLE IF NOT EXISTS ods.order_snapshot_dist ON CLUSTER clickstream_cluster -AS ods.order_snapshot_rep -ENGINE = Distributed('clickstream_cluster', 'ods', 'order_snapshot_rep', cityHash64(order_id)); +CREATE TABLE IF NOT EXISTS ods.order_dist ON CLUSTER clickstream_cluster +AS ods.order_rep +ENGINE = Distributed('clickstream_cluster', 'ods', 'order_rep', cityHash64(order_id)); -- Локальная таблица брака слепка. -- @@ -215,7 +215,7 @@ ENGINE = Distributed('clickstream_cluster', 'ods', 'order_snapshot_rep', cityHas -- field_invalid. Имя провалившегося поля в класс не входит — сырой текст лежит -- рядом, и единичный случай разбирается по нему, без постоянной детализации -- предиката (docs/architecture/orders/ingestion.md). -CREATE TABLE IF NOT EXISTS ods.order_snapshot_errors_rep ON CLUSTER clickstream_cluster +CREATE TABLE IF NOT EXISTS ods.order_errors_rep ON CLUSTER clickstream_cluster ( raw String, error_class LowCardinality(String), @@ -233,6 +233,6 @@ ORDER BY (error_class, kafka_partition, kafka_offset) TTL toDateTime(_load_ts) + INTERVAL 1 MONTH SETTINGS ttl_only_drop_parts = 1; -CREATE TABLE IF NOT EXISTS ods.order_snapshot_errors_dist ON CLUSTER clickstream_cluster -AS ods.order_snapshot_errors_rep -ENGINE = Distributed('clickstream_cluster', 'ods', 'order_snapshot_errors_rep', cityHash64(raw)); +CREATE TABLE IF NOT EXISTS ods.order_errors_dist ON CLUSTER clickstream_cluster +AS ods.order_errors_rep +ENGINE = Distributed('clickstream_cluster', 'ods', 'order_errors_rep', cityHash64(raw)); diff --git a/sql/ddl/30-ods-views.sql b/sql/ddl/30-ods-views.sql index a5d7f29..51ae10c 100644 --- a/sql/ddl/30-ods-views.sql +++ b/sql/ddl/30-ods-views.sql @@ -22,7 +22,7 @@ FROM ods.event_dist FINAL; CREATE VIEW IF NOT EXISTS ods.order_v ON CLUSTER clickstream_cluster AS SELECT * -FROM ods.order_snapshot_dist FINAL; +FROM ods.order_dist FINAL; -- ODS: разбор сырья в событие и в таблицу ошибок. -- diff --git a/sql/ods/order_snapshot_errors_load.sql b/sql/ods/order_errors_load.sql similarity index 95% rename from sql/ods/order_snapshot_errors_load.sql rename to sql/ods/order_errors_load.sql index 597e9ab..eaac128 100644 --- a/sql/ods/order_snapshot_errors_load.sql +++ b/sql/ods/order_errors_load.sql @@ -4,7 +4,7 @@ -- первый совпавший: скаляр проваливает и проверку на объект, и сверку ключей — -- без объявленного порядка он попал бы то в один класс, то в другой. -INSERT INTO ods.order_snapshot_errors_dist +INSERT INTO ods.order_errors_dist {% include "ods/_order_wire_contract.sql" %} SELECT raw, diff --git a/sql/ods/order_snapshot_load.sql b/sql/ods/order_load.sql similarity index 98% rename from sql/ods/order_snapshot_load.sql rename to sql/ods/order_load.sql index 5c5f8a9..5b3612c 100644 --- a/sql/ods/order_snapshot_load.sql +++ b/sql/ods/order_load.sql @@ -4,7 +4,7 @@ -- Разобранные значения берёт assumeNotNull: обнуляемый разбор стоит за -- предикатом, который NULL уже отсёк, и приведение здесь не может упасть. -INSERT INTO ods.order_snapshot_dist +INSERT INTO ods.order_dist {% include "ods/_order_wire_contract.sql" %} SELECT JSONExtractString(raw, 'order_id') AS order_id,