diff --git a/Makefile b/Makefile index bd671c1..326064e 100644 --- a/Makefile +++ b/Makefile @@ -224,10 +224,14 @@ test: lint: $(UV) run black --check airflow tests $(UV) run isort --check-only airflow tests + $(UV) run black --check scripts/check_airports_snapshot.py + $(UV) run isort --check-only scripts/check_airports_snapshot.py fmt: $(UV) run black airflow tests $(UV) run isort airflow tests + $(UV) run black scripts/check_airports_snapshot.py + $(UV) run isort scripts/check_airports_snapshot.py clean-venv: python -c "import shutil; shutil.rmtree('.venv', ignore_errors=True)" diff --git a/README.md b/README.md index eacee59..9af25cc 100644 --- a/README.md +++ b/README.md @@ -104,9 +104,9 @@ SELECT COUNT(*) FROM dm.sales_report; оно проведет от билетов и сегментов к факту и витрине. Рядом держите [интерактивную карту](docs/design/architecture-map.html) — локальный файл `docs/design/architecture-map.html` нужно открыть в браузере; в Gitea HTML -показывается как исходный текст. Затем разберите -[готовую загрузку справочника](docs/design/reading_the_pipeline.md#ods-airports) -и переходите к [первому заданию ODS](docs/assignment/README.md#первая-загрузка-ods). +показывается как исходный текст. Затем переходите к +[учебным заданиям](docs/assignment/README.md): начните с приема справочника +через PXF, после него — загрузки ODS. ## DAG-и в стенде @@ -115,6 +115,7 @@ SELECT COUNT(*) FROM dm.sales_report; - `bookings_stg_ddl` — создаёт/обновляет весь STG слой для bookings (9 таблиц: bookings, tickets, airports, airplanes, routes, seats, flights, segments, boarding_passes; включая внешние `*_ext` через PXF); - `bookings_to_gp_stage` — генерирует учебный день в `bookings-db`, затем загружает данные в STG и выполняет DQ‑проверки. +- `lab_pxf_airports` — готовит общий снимок четырех справочников для [лабораторной PXF](docs/assignment/pxf_airports.md); использует те же SQL и DQ, без генерации транзакций. - `bookings_ods_ddl` — создаёт/обновляет ODS-таблицы по домену bookings. - `bookings_to_gp_ods` — заменяет снимки справочников и обновляет транзакционные таблицы ODS из STG; выполняет DQ-проверки. - `bookings_dds_ddl` — создаёт/обновляет DDS-таблицы (`dim_*`, `fact_flight_sales`) по домену bookings. diff --git a/airflow/dags/lab_pxf_airports.py b/airflow/dags/lab_pxf_airports.py new file mode 100644 index 0000000..373c9a6 --- /dev/null +++ b/airflow/dags/lab_pxf_airports.py @@ -0,0 +1,70 @@ +"""Готовый DAG первой лабораторной: общий снимок справочников без генерации дней.""" + +from datetime import timedelta + +import pendulum +from airflow.providers.postgres.operators.postgres import PostgresOperator + +from airflow import DAG + +with DAG( + dag_id="lab_pxf_airports", + start_date=pendulum.datetime(2017, 1, 1, tz="UTC"), + schedule=None, + catchup=False, + max_active_runs=1, + template_searchpath="/sql", + default_args={ + "owner": "airflow", + "retries": 1, + "retry_delay": timedelta(seconds=30), + }, + tags=["lab", "bookings", "pxf"], + description="Лабораторная PXF: четыре снимка STG с общим Run ID для ODS", +) as dag: + load_airports_to_stg = PostgresOperator( + task_id="load_airports_to_stg", + postgres_conn_id="greenplum_conn", + sql="stg/airports_load.sql", + ) + check_airports_dq = PostgresOperator( + task_id="check_airports_dq", + postgres_conn_id="greenplum_conn", + sql="stg/airports_dq.sql", + ) + load_airplanes_to_stg = PostgresOperator( + task_id="load_airplanes_to_stg", + postgres_conn_id="greenplum_conn", + sql="stg/airplanes_load.sql", + ) + check_airplanes_dq = PostgresOperator( + task_id="check_airplanes_dq", + postgres_conn_id="greenplum_conn", + sql="stg/airplanes_dq.sql", + ) + load_routes_to_stg = PostgresOperator( + task_id="load_routes_to_stg", + postgres_conn_id="greenplum_conn", + sql="stg/routes_load.sql", + ) + check_routes_dq = PostgresOperator( + task_id="check_routes_dq", + postgres_conn_id="greenplum_conn", + sql="stg/routes_dq.sql", + ) + load_seats_to_stg = PostgresOperator( + task_id="load_seats_to_stg", + postgres_conn_id="greenplum_conn", + sql="stg/seats_load.sql", + ) + check_seats_dq = PostgresOperator( + task_id="check_seats_dq", + postgres_conn_id="greenplum_conn", + sql="stg/seats_dq.sql", + ) + + load_airports_to_stg >> check_airports_dq + load_airplanes_to_stg >> check_airplanes_dq + # Готовим один общий батч: DQ маршрутов и мест читает связанные справочники. + [check_airports_dq, check_airplanes_dq] >> load_routes_to_stg >> check_routes_dq + check_airplanes_dq >> load_seats_to_stg >> check_seats_dq diff --git a/docs/README.md b/docs/README.md index bf7616b..2624129 100644 --- a/docs/README.md +++ b/docs/README.md @@ -2,14 +2,15 @@ ## Знакомство и первое задание -После [запуска стенда](../README.md) пройдите три шага: +После [запуска стенда](../README.md): 1. [Структура DWH](design/db_schema.md): смысл строк и связей на примере поездки. -2. [От карты к SQL загрузки](design/reading_the_pipeline.md#ods-airports): - разбор готового справочника ODS. Сборка факта понадобится позже, перед DDS. -3. [Учебные задания](assignment/README.md): файлы для работы, порядок запуска - и [самопроверка](assignment/README.md#запуск-и-проверка). - Поля и бизнес-правила находятся в [ТЗ](assignment/analyst_spec.md). +2. [Учебные задания](assignment/README.md): прием справочника через PXF, + затем загрузки ODS, измерения и витрины. Каждая работа указывает свои проверки. + +Перед заданиями ODS прочитайте [разбор готового справочника](design/reading_the_pipeline.md#ods-airports). +Сборка факта из того же руководства понадобится перед DDS. +Поля и бизнес-правила расширений находятся в [ТЗ](assignment/analyst_spec.md). ## Найти деталь во время работы diff --git a/docs/assignment/README.md b/docs/assignment/README.md index 1479c9c..bbc33da 100644 --- a/docs/assignment/README.md +++ b/docs/assignment/README.md @@ -1,8 +1,9 @@ # Учебные задания В стенде уже работает цепочка от Bookings до `dm.sales_report`. -Вы дополните ее загрузками ODS, измерениями DDS и четырьмя витринами DM. -Поля, преобразования и бизнес-правила описаны в [ТЗ](analyst_spec.md). +Сначала вы восстановите прием справочника в STG, затем дополните цепочку +загрузками ODS, измерениями DDS и четырьмя витринами DM. +Требования к этим расширениям описаны в [ТЗ](analyst_spec.md). ## Подготовка @@ -12,23 +13,31 @@ Открывайте [карту](../design/architecture-map.html) рядом с заданием; для Gitea есть [инструкция открытия локального HTML](../design/db_schema.md#как-открыть-карту). -Таблицы уже созданы, задачи Airflow подключены к SQL-файлам. -В учебных `*_load.sql` и `*_dq.sql` пока стоит `SELECT 1;`. -Вы заменяете эти заглушки; менять DAG и создавать таблицы заново не требуется. +## Прием справочника: PXF → STG + +Пройдите [«Восстановите прием справочника из источника»](pxf_airports.md). +Вы разберете готовый участок `airports`, восстановите внешнюю таблицу +и загрузку. На повторном запуске и переименовании аэропорта выясните, +как загрузка сохраняет снимки источника. ## Первая загрузка: ODS -Начните с `ods.airplanes`, затем добавьте `ods.seats`. +После приема справочника реализуйте `ods.airplanes`, затем `ods.seats`. На карте откройте [модели самолетов](../design/architecture-map.html#node=ods.airplanes) и [места](../design/architecture-map.html#node=ods.seats): источники STG уже заполнены. Позже обе таблицы понадобятся `dds.dim_airplanes` для характеристик модели и расчета ее вместимости. +Таблицы уже созданы, задачи Airflow подключены к SQL-файлам. +В учебных `*_load.sql` и `*_dq.sql` пока стоит `SELECT 1;`. +В заданиях ODS, DDS и DM вы заменяете эти заглушки; менять DAG +и создавать таблицы заново не требуется. + ### Готовый пример Пройдите [разбор stg.airports → ods.airports](../design/reading_the_pipeline.md#ods-airports). Это близкий пример полного снимка с выбором батча, преобразованиями и дедупликацией. -Для первого задания этого разбора достаточно. +Для первых загрузок ODS этого разбора достаточно. ### Самостоятельная работа @@ -46,8 +55,8 @@ ### Запуск и проверка -1. Дождитесь успешного `bookings_to_gp_stage` из подготовки. - Если STG уже заполнен, новый запуск для этой работы не нужен. +1. Используйте общий снимок после восстановления источника + в лабораторной PXF. Новый запуск STG для этой работы не нужен. 2. Запустите `bookings_to_gp_ods` целиком. Он выберет согласованный снимок в `resolve_stg_batch_id`, затем выполнит `load_ods_airplanes → dq_ods_airplanes → load_ods_seats → dq_ods_seats`. diff --git a/docs/assignment/analyst_spec.md b/docs/assignment/analyst_spec.md index cd0daba..6e5f0df 100644 --- a/docs/assignment/analyst_spec.md +++ b/docs/assignment/analyst_spec.md @@ -13,8 +13,9 @@ > > SQL-скрипты эталона лежат в `sql/stg/`, `sql/ods/`, `sql/dds/`, `sql/dm/`. -Порядок запуска и проверки первой загрузки находится во -[входе в задания](README.md#первая-загрузка-ods). В каждом разделе ниже есть +Программа начинается с [приема справочника через PXF](pxf_airports.md). +Затем переходите к расширениям из этого ТЗ; порядок их запуска и проверки +находится во [входе в задания](README.md#первая-загрузка-ods). В каждом разделе ниже есть ссылка на объект карты и близкий готовый пример. Карту открывайте [локально в браузере](../design/db_schema.md#как-открыть-карту). @@ -23,7 +24,8 @@ ## Рекомендуемый порядок выполнения > **Примечание:** STG-слой полностью реализован в эталоне (все 9 таблиц), `ods.routes` -> тоже эталонный. Ваше задание начинается с ODS: `airplanes` и `seats`. +> тоже эталонный. В первой лабораторной вы временно разбираете и восстанавливаете +> прием `airports`; расширения по этому ТЗ начинаются с ODS: `airplanes` и `seats`. 1. **ODS** — `airplanes`, `seats` (практика TRUNCATE+INSERT по аналогии с `ods.airports`) 2. **DDS** — `dim_airplanes`, `dim_passengers` (SCD1 — новые измерения) @@ -75,7 +77,7 @@ STG-слой (все 9 таблиц) и `ods.routes` полностью реал `ods.routes` тоже эталонный: факт `fact_flight_sales` использует его для резолвинга аэропортов вылета/прилёта, независимо от студенческого `dds.dim_routes`. -Для первого задания разберите [загрузку `ods.airports`](../design/reading_the_pipeline.md#ods-airports). +После лабораторной PXF разберите [загрузку `ods.airports`](../design/reading_the_pipeline.md#ods-airports). При необходимости проследить получение снимка откройте [sql/stg/airports_load.sql](../../sql/stg/airports_load.sql) или [sql/stg/airplanes_load.sql](../../sql/stg/airplanes_load.sql): оба читают полный diff --git a/docs/assignment/pxf_airports.md b/docs/assignment/pxf_airports.md new file mode 100644 index 0000000..6e0ba01 --- /dev/null +++ b/docs/assignment/pxf_airports.md @@ -0,0 +1,317 @@ +# Восстановите прием справочника из источника + +Хранилище получает справочник аэропортов из Postgres. Названия и другие +атрибуты в источнике могут меняться, поэтому загрузка должна сохранять +состояние справочника на момент приема данных. + +Вы разберете готовый участок `airports` и напишете его заново: создадите +внешнюю таблицу PXF и загрузку полного снимка в STG. Затем испытаете ее +на повторном запуске и изменении источника. Полученный снимок понадобится +следующему слою — ODS. + +## Задача + +Ваш участок на [карте](../design/architecture-map.html#node=stg.airports): +`bookings.airports_data → stg.airports_ext → stg.airports`. + +| Что написать | Где | +|---|---| +| DDL внешней таблицы `stg.airports_ext` | [airports_ddl.sql](../../sql/stg/airports_ddl.sql) | +| Загрузку полного снимка во внутреннюю `stg.airports` | [airports_load.sql](../../sql/stg/airports_load.sql) | + +Внутренняя таблица, JDBC-сервер PXF, DAG и проверки качества уже готовы. +Рядом есть пример для другого справочника: +[DDL airplanes](../../sql/stg/airplanes_ddl.sql) и +[его загрузка](../../sql/stg/airplanes_load.sql). Используйте его, учитывая +состав полей и ключ аэропорта. + +Требования к вашей загрузке: + +- Источник — `bookings.airports_data` в базе `demo`, ключ — `airport_code`. +- Все шесть полей (`airport_code`, `airport_name`, `city`, `country`, + `coordinates`, `timezone`) попадают в STG без потерь и преобразований. + Внешняя и внутренняя таблицы используют TEXT; JSON и координаты + пока хранятся как текст. +- Строки одного снимка получают `_load_id = '{{ run_id }}'`, + `_load_ts = now()` и `event_ts = now()`. Точного времени изменения + аэропорта в источнике нет, поэтому `event_ts` — время получения снимка. +- Новый запуск сохраняет новый полный снимок, оставляя предыдущие. + Повторное выполнение загрузки в том же запуске на неизменном источнике + не меняет бизнес-данные и не создает дубли ключей внутри снимка. + +## Подготовка + +Нужен свой стенд после [быстрого старта и знакомства с моделью](README.md#подготовка). +Во время работы не запускайте другие загрузки: они могут изменить данные +посреди опыта. Если запущен `bookings_validate`, дождитесь его завершения. + +Сначала освободите участок для своей реализации. Команды ниже удаляют +внешнюю таблицу и **всю историю только `stg.airports`**. Это нужно, чтобы +данные быстрого старта не скрыли отсутствие вашей загрузки. + +
+Подготовка файлов и таблиц + +Сохраните свои SQL-файлы в Git или отдельной копии. В `airports_ddl.sql` +удалите блок от `DROP EXTERNAL TABLE` до конца `CREATE EXTERNAL TABLE`; +DDL схемы и внутренней таблицы оставьте. Содержимое `airports_load.sql` +замените на `SELECT 1;`. DAG и DQ менять не нужно. + +В Greenplum (`make gp-psql`): + +```sql +DROP EXTERNAL TABLE IF EXISTS stg.airports_ext; +TRUNCATE TABLE stg.airports; +SELECT COUNT(*) FROM stg.airports; -- 0 +``` + +Остальные справочники и текущий ODS сохраняются. Пока участок +не восстановлен, ODS запускать не нужно. + +
+ +## Соберите загрузку + +Начните с внешней таблицы. Добейтесь, чтобы запрос к `stg.airports_ext` +возвращал те же поля и значения, что запрос к `bookings.airports_data` +в Postgres. Для начала достаточно рассмотреть несколько аэропортов. + +
+Ориентиры для PXF и применение DDL + +В `LOCATION` укажите таблицу источника, `PROFILE=JDBC` и `SERVER=bookings-db`. +`SERVER` — имя готового [каталога конфигурации PXF](../../pxf/servers/bookings-db/jdbc-site.xml). +Формат чтения — `FORMAT 'CUSTOM' (formatter='pxfwritable_import')`. +Логин и пароль в DDL не нужны. Если меняли учетные данные источника, +согласуйте их с PXF по [справочнику стенда](../stack.md). + +Примените свой файл из `make gp-psql`: + +```sql +\set ON_ERROR_STOP on +\i /sql/stg/airports_ddl.sql +SELECT * FROM stg.airports_ext ORDER BY airport_code LIMIT 3; +``` + +В Postgres (`make bookings-psql`) выполните такой же запрос к +`bookings.airports_data`. Сверьте состав и порядок полей с DDL. +Если чтение не работает, начните с сообщения об ошибке и +[диагностики STG](../bookings_to_gp_stage.md#если-загрузка-не-прошла). + +
+ +Теперь напишите загрузку снимка. Продумайте, как ваш SQL отличит +повторное выполнение от нового запуска и сохранит историю. +Выполнять загрузку будет готовый DAG `lab_pxf_airports`. + +Получите первый снимок и сравните его с источником готовой сверкой. +Она проверяет ключи, все бизнес-поля и служебные метки. Зеленые задачи +в Airflow показывают, что код выполнился; сверка помогает обнаружить +ошибку в самих данных. + +
+Первый запуск в Airflow и сверка с Postgres + +Для сверки нужен `uv` с Python 3.11 из +[окружения разработчика](../stack.md#локальное-окружение-разработчика-uv). +В терминале из корня репозитория сохраните исходный справочник: + +```bash +mkdir -p data/lab9 +uv run --no-project scripts/check_airports_snapshot.py export data/lab9/airports-before.json +``` + +Скрипт читает Postgres напрямую, минуя вашу внешнюю таблицу, и не перезаписывает +существующий файл. Эта копия позволит сравнить первый снимок с источником +даже после изменения названия аэропорта. + +Откройте Airflow UI по адресу из `.env` (`AIRFLOW_WEB_PORT`), войдите с +`AIRFLOW_USER` и `AIRFLOW_PASSWORD`. Включите `lab_pxf_airports`, нажмите +**Trigger DAG** и дождитесь успеха всех задач. В **Grid** выберите запуск; +его **Run ID** подставьте в команду: + +```bash +first_run='вставьте Run ID первого запуска' +uv run --no-project scripts/check_airports_snapshot.py check "$first_run" data/lab9/airports-before.json +``` + +Если задача упала, откройте ее **Log**. SQL-файлы доступны контейнеру +из рабочей папки: после правки пересобирать образ не нужно. +`{{ run_id }}` подставляет Airflow, поэтому файл загрузки не следует +выполнять напрямую через `psql`. + +
+ +## Испытайте загрузку + +### Повторное выполнение + +Представьте, что после сбоя пришлось повторить уже выполненную задачу. +Что произойдет с данными при повторном выполнении вашего SQL? За счет +чего в одном снимке не появится вторая копия каждого аэропорта? + +Повторите загрузку **в том же запуске** и снова выполните сверку первого +снимка. При ошибке исправьте SQL и повторите опыт на том же Run ID. + +
+Как повторить задачу, сохранив Run ID + +В запуске первого снимка выберите `load_airports_to_stg` → **Clear task**. +Снимите **Upstream**, **Downstream**, **Past**, **Future**, **Recursive** +и **Only Failed**, если включены. В **Affected Tasks** должна остаться +только эта задача этого запуска. Нажмите **Clear**, дождитесь успеха, +затем так же повторите `check_airports_dq`. + +Новый **Trigger DAG** создал бы другой Run ID. Здесь нужно повторное +выполнение с прежним `_load_id`, чтобы испытать защиту от дублей. + +Если загрузка уже записала дубли или неверные значения, исправление SQL +само их не удалит. +Удалите из `stg.airports` строки только этого `_load_id`, затем повторите +исправленную загрузку и DQ в том же запуске. После успешной сверки повторите +их еще раз: теперь вы проверяете защиту от дублей на заполненном снимке. + +
+ +### Изменение в источнике + +Теперь аэропорт переименовали. До новой загрузки предположите, где будет +видно новое название: в Postgres, во внешней таблице, в первом снимке STG? + +Измените русское название одного аэропорта, сохранив ключ и остальные +языки. Для опыта ниже есть готовый SQL, который также сохраняет исходное +название для восстановления. + +
+Изменить название в Postgres + +В `make bookings-psql`: + +```sql +\set ON_ERROR_STOP on +CREATE TABLE public.lab9_airport_original AS +SELECT airport_code, airport_name +FROM bookings.airports_data ORDER BY airport_code LIMIT 1; + +UPDATE bookings.airports_data AS a +SET airport_name = jsonb_set( + b.airport_name, '{ru}', to_jsonb((b.airport_name->>'ru') || ' [lab9]'), false +) +FROM public.lab9_airport_original AS b +WHERE a.airport_code = b.airport_code +RETURNING a.airport_code, b.airport_name AS original, a.airport_name AS changed; +``` + +Запрос изменяет один аэропорт. Если таблица копии уже существует после +предыдущего опыта, сначала [восстановите источник](#завершение-работы), +сверьте результат и удалите таблицу копии. Затем можно начать новый опыт. + +
+ +Найдите этот аэропорт в `stg.airports_ext` и `stg.airports` обычными +SELECT-запросами. Сопоставьте наблюдение со своим предположением. + +Запустите `lab_pxf_airports` заново и рассмотрите оба снимка в STG. +Какое название хранится в каждом? Почему одного сравнения количества +строк недостаточно, чтобы заметить переименование? + +Сверьте новый снимок с измененным источником, а первый — с исходным. +Это обнаружит, например, загрузку старых данных под новым Run ID. + +
+Сверить оба снимка + +В том же терминале, где задан `first_run`: + +```bash +uv run --no-project scripts/check_airports_snapshot.py export data/lab9/airports-after.json +next_run='вставьте Run ID нового запуска' +uv run --no-project scripts/check_airports_snapshot.py check "$next_run" data/lab9/airports-after.json +uv run --no-project scripts/check_airports_snapshot.py check "$first_run" data/lab9/airports-before.json +``` + +Если терминал закрывался, значения Run ID можно снова взять из Airflow. + +
+ +## Передайте снимок в ODS + +Запустите `bookings_to_gp_ods` без параметров `conf`. Найдите измененный +аэропорт в `ods.airports`: название должно быть новым, а `_load_id` — +совпадать с последним запуском STG. Так вы увидите, что следующий слой +использовал результат вашей загрузки. + +ODS выбирает общий батч четырех справочников: `airports`, `airplanes`, +`routes`, `seats`. Поэтому готовый `lab_pxf_airports` загружает все четыре, +используя те же SQL, что основной STG-DAG. Генератор новых дней и +транзакционные таблицы в этом опыте не участвуют. + +
+Проверить выбор снимка и данные ODS + +В Log задачи `resolve_stg_batch_id` строка `Используем stg_batch_id=...` +показывает выбранный Run ID. Он должен совпасть с `next_run`. +После завершения DAG выполните: + +```bash +uv run --no-project scripts/check_airports_snapshot.py check "$next_run" data/lab9/airports-after.json --layer ods +``` + +Сверка учитывает преобразование JSON в русские названия и требует нужный +`_load_id` у всех строк ODS. Если выбран старый батч, посмотрите, завершились +ли все загрузки и DQ последнего запуска `lab_pxf_airports`. + +
+ +## Завершение работы + +Верните исходное название и получите еще один общий снимок через +`lab_pxf_airports`. Проведите его по цепочке +`bookings_to_gp_ods` → `bookings_to_gp_dds` → `bookings_to_gp_dm`. +В `ods.airports` и `dds.dim_airports` должно вернуться исходное название. + +
+Восстановить источник и сверить результат + +В `make bookings-psql`: + +```sql +UPDATE bookings.airports_data AS a +SET airport_name = b.airport_name +FROM public.lab9_airport_original AS b +WHERE a.airport_code = b.airport_code +RETURNING a.airport_code, a.airport_name; +``` + +После нового запуска `lab_pxf_airports` и ODS сравните восстановленный +снимок с исходной копией: + +```bash +restored_run='вставьте Run ID запуска STG после восстановления' +uv run --no-project scripts/check_airports_snapshot.py check "$restored_run" data/lab9/airports-before.json +uv run --no-project scripts/check_airports_snapshot.py check "$restored_run" data/lab9/airports-before.json --layer ods +``` + +Повтор UPDATE восстановления безопасен. После успешной сверки выполните +в Postgres `DROP TABLE public.lab9_airport_original;`. +Если прервались на изменении источника, +продолжите с этого восстановления и нового запуска цепочки. + +Для повторения всей лабораторной возьмите новые имена контрольных файлов. +При полной пересборке сохраните SQL-файлы: `make clean` удаляет тома. +Затем снова пройдите быстрый старт со своим решением. + +
+ +Снимки в STG остаются историей опыта: первоначальный, с измененным названием +и восстановленный. Последний общий снимок и текущий ODS снова содержат +исходные значения; транзакционные данные остались от быстрого старта. + +Результат работы — ваши два SQL-файла и объяснение их поведения: откуда +берется каждый снимок, что защищает его от дублей при повторе и как +изменение источника попадает в следующую загрузку. + +Дальше — [загрузки `ods.airplanes` и `ods.seats`](README.md#первая-загрузка-ods). +Они используют уже подготовленный снимок, поэтому заново запускать STG +для начала следующей работы не нужно. Там вы будете преобразовывать поля +полученных данных и заменять текущее состояние в ODS. diff --git a/docs/bookings_to_gp_ods.md b/docs/bookings_to_gp_ods.md index 321a54a..3edc4bc 100644 --- a/docs/bookings_to_gp_ods.md +++ b/docs/bookings_to_gp_ods.md @@ -124,7 +124,7 @@ STG-батч. Сравните его с XCom задачи `resolve_stg_batch_id Пустой транзакционный батч допустим, пустой снимок справочника считается ошибкой. После реализации `airplanes` и `seats` пройдите -[самопроверку первого задания](assignment/README.md#запуск-и-проверка). +[самопроверку заданий ODS](assignment/README.md#запуск-и-проверка). Дальше можно переходить к [DDS](bookings_to_gp_dds.md). ## Если загрузка не прошла diff --git a/docs/bookings_to_gp_stage.md b/docs/bookings_to_gp_stage.md index eb3f8e3..637bed5 100644 --- a/docs/bookings_to_gp_stage.md +++ b/docs/bookings_to_gp_stage.md @@ -1,7 +1,9 @@ # Загрузка Bookings → STG: `bookings_to_gp_stage` DAG читает девять таблиц Bookings через внешние таблицы PXF и сохраняет -строки в Greenplum. Весь STG, включая DQ, уже реализован на `main`. +строки в Greenplum. Весь STG, включая DQ, уже реализован на `main`; +в [первой лабораторной](assignment/pxf_airports.md) вы разберете и восстановите +прием `airports`. Смысл бронирования, билета и сегмента разобран в [руководстве о структуре](design/db_schema.md); на карте начните со [stg.bookings](design/architecture-map.html#node=stg.bookings). diff --git a/docs/dag_execution_order.md b/docs/dag_execution_order.md index 8ec3890..44631e2 100644 --- a/docs/dag_execution_order.md +++ b/docs/dag_execution_order.md @@ -54,3 +54,9 @@ DDL создает структуру; для наполнения нужны о `bookings_validate` запускайте отдельно от загрузок: дождитесь всего DAG и успешного `validate_dds.scd2_restore` перед следующей записью в хранилище. Тест SCD2 временно меняет данные и восстанавливает их из копии. + +В [лабораторной PXF](assignment/pxf_airports.md) снимки справочников готовит +`lab_pxf_airports`: он выполняет только четыре справочника STG и их DQ. +Для передачи снимка и восстановления нижних слоев используйте обычные +DAG ODS → DDS → DM. Следующий учебный день по-прежнему загружается полной +цепочкой, начиная с `bookings_to_gp_stage`. diff --git a/docs/design/architecture-map.html b/docs/design/architecture-map.html index 0235035..bcec434 100644 --- a/docs/design/architecture-map.html +++ b/docs/design/architecture-map.html @@ -1222,6 +1222,10 @@ { "label": "DQ", "url": "../../sql/stg/airports_dq.sql" + }, + { + "label": "Лабораторная: прием через PXF", + "url": "../assignment/pxf_airports.md" } ], "parts": [ diff --git a/docs/design/architecture-map/map.json b/docs/design/architecture-map/map.json index ca0c341..aa409e4 100644 --- a/docs/design/architecture-map/map.json +++ b/docs/design/architecture-map/map.json @@ -1022,6 +1022,10 @@ { "label": "DQ", "url": "../../sql/stg/airports_dq.sql" + }, + { + "label": "Лабораторная: прием через PXF", + "url": "../assignment/pxf_airports.md" } ], "parts": [ diff --git a/docs/design/db_schema.md b/docs/design/db_schema.md index 92ae928..edc3ba3 100644 --- a/docs/design/db_schema.md +++ b/docs/design/db_schema.md @@ -171,7 +171,7 @@ SQL использует `LEFT JOIN`, поэтому строка с таким эталон работает до выполнения заданий. Остальные витрины видны рядом пунктирными рамками: в их карточках описан требуемый результат по ТЗ. -## Модели самолетов и места: начало задания +## Модели самолетов и места: расширение цепочки Откройте [dds.dim_airplanes](architecture-map.html#node=dds.dim_airplanes). У него два источника: `ods.airplanes` дает характеристики модели, @@ -203,13 +203,13 @@ SQL использует `LEFT JOIN`, поэтому строка с таким ## Куда идти дальше -Для первого задания разберите готовую загрузку справочника: -[от карты к SQL загрузки](reading_the_pipeline.md#ods-airports), затем откройте -[первое задание ODS](../assignment/README.md#первая-загрузка-ods). Для точечного вопроса +Начните с [приема справочника через PXF](../assignment/pxf_airports.md), +затем переходите к [заданиям ODS](../assignment/README.md#первая-загрузка-ods). +Для точечного вопроса о таблице возвращайтесь на карту: там находится каталог объектов, их источников, потребителей, SQL и DQ. -Проверяйте задание по [инструкции самопроверки](../assignment/README.md#запуск-и-проверка). +Проверки каждого этапа находятся в [учебном маршруте](../assignment/README.md). Для разбора ошибки загрузки откройте руководство нужного слоя: [STG](../bookings_to_gp_stage.md), [ODS](../bookings_to_gp_ods.md), [DDS](../bookings_to_gp_dds.md), [DM](../bookings_to_gp_dm.md). diff --git a/docs/design/reading_the_pipeline.md b/docs/design/reading_the_pipeline.md index d5d7ea2..58e05af 100644 --- a/docs/design/reading_the_pipeline.md +++ b/docs/design/reading_the_pipeline.md @@ -5,7 +5,7 @@ что меняет и по какому ключу записывает результат. Смысл строк и связей разобран в [руководстве о структуре DWH](db_schema.md). -Для первого задания достаточно примера `ods.airports` ниже. +Для заданий на снимки ODS достаточно примера `ods.airports` ниже. К сборке факта вернитесь перед заданиями DDS. Адресные ссылки на карту работают в локальном просмотре; если читаете в Gitea, используйте [инструкцию открытия карты](db_schema.md#как-открыть-карту). diff --git a/scripts/check_airports_snapshot.py b/scripts/check_airports_snapshot.py new file mode 100644 index 0000000..8441cdf --- /dev/null +++ b/scripts/check_airports_snapshot.py @@ -0,0 +1,158 @@ +"""Независимая сверка лабораторной PXF: Postgres → сохраненный файл → STG/ODS.""" + +from __future__ import annotations + +import argparse +import json +import subprocess +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +FIELDS = ("airport_code", "airport_name", "city", "country", "coordinates", "timezone") +JSON_FIELDS = {"airport_name", "city", "country"} + + +def read_rows(service: str, query: str) -> list[dict]: + """Читает через psql в контейнере; настройки берет из окружения Compose.""" + if service == "bookings-db": + command = ( + 'PGPASSWORD="$POSTGRES_PASSWORD" exec psql -X -qAt ' + '-v ON_ERROR_STOP=1 -h 127.0.0.1 -U "$POSTGRES_USER" -d demo' + ) + else: + command = ( + 'PGPASSWORD="$GREENPLUM_PASSWORD" exec ' + "/usr/local/greenplum-db/bin/psql -X -qAt -v ON_ERROR_STOP=1 " + '-h 127.0.0.1 -U "$GREENPLUM_USER" -d "$GREENPLUM_DATABASE_NAME"' + ) + result = subprocess.run( + ["docker", "compose", "exec", "-T", service, "sh", "-c", command], + input=f"SELECT row_to_json(r) FROM ({query}) r;\n", + text=True, + capture_output=True, + cwd=ROOT, + check=True, + ) + return [json.loads(line) for line in result.stdout.splitlines() if line.strip()] + + +def compare_snapshot( + expected: list[dict], actual: list[dict], batch_id: str, layer: str +) -> list[str]: + """Проверяет ключи, все бизнес-поля и метки; порядок строк не важен.""" + errors = [] + if not expected: + return ["Контрольный источник пуст. Сначала восстановите Bookings."] + source = {row["airport_code"]: row for row in expected} + if len(source) != len(expected) or None in source or "" in source: + return ["В контрольном источнике пустые или повторяющиеся ключи."] + keys = [row["airport_code"] for row in actual] + if len(keys) != len(set(keys)): + errors.append("Повторяются airport_code внутри снимка.") + missing = source.keys() - set(keys) + extra = set(keys) - source.keys() + if missing: + errors.append( + f"Не получены ключи: {sorted(missing)[:5]} (всего {len(missing)})." + ) + if extra: + errors.append(f"Лишние ключи: {sorted(map(str, extra))[:5]}.") + for row in actual: + key = row["airport_code"] + if row.get("_load_id") != batch_id: + errors.append( + f"{key}: _load_id не равен {batch_id}; проверьте выбор батча." + ) + timestamps = ("_load_ts", "event_ts") if layer == "stg" else ("_load_ts",) + if any(not row.get(field) for field in timestamps): + errors.append(f"{key}: не заполнены метки времени {timestamps}.") + if key not in source: + continue + for field in FIELDS: + wanted, received = source[key][field], row[field] + if field in JSON_FIELDS: + # Сравниваем весь JSON в STG, включая языки, кроме русского. + wanted = json.loads(wanted) if wanted is not None else None + if layer == "ods": + wanted = wanted.get("ru") if wanted is not None else None + elif received is not None: + try: + received = json.loads(received) + except (ValueError, TypeError): + errors.append(f"{key}: {field} не содержит исходный JSON.") + continue + if received != wanted: + errors.append(f"{key}: отличается {field}.") + return errors + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + actions = parser.add_subparsers(dest="action", required=True) + export = actions.add_parser( + "export", help="Сохранить источник напрямую из Postgres" + ) + export.add_argument( + "file", type=Path, help="Новый JSON-файл для контрольного источника" + ) + check = actions.add_parser("check", help="Сверить снимок с сохраненным источником") + check.add_argument("batch_id", help="Run ID нужного запуска STG") + check.add_argument("file", type=Path, help="Контрольный JSON-файл из export") + check.add_argument("--layer", choices=("stg", "ods"), default="stg") + args = parser.parse_args() + + try: + if args.action == "export": + fields = ", ".join(f"{field}::text AS {field}" for field in FIELDS) + rows = read_rows( + "bookings-db", + f"SELECT {fields} FROM bookings.airports_data ORDER BY airport_code", + ) + if not rows: + raise ValueError( + "Источник пуст. Пройдите инициализацию Bookings из README." + ) + rows.sort(key=lambda row: row["airport_code"]) + # Не даем случайно заменить контроль A данными измененного источника B. + with args.file.open("x", encoding="utf-8") as output: + json.dump(rows, output, ensure_ascii=False, indent=2) + output.write("\n") + print(f"Источник: {len(rows)} аэропортов сохранено в {args.file}.") + return 0 + + expected = json.loads(args.file.read_text(encoding="utf-8")) + # Run ID передается SQL-литералом, а не частью команды shell. + batch_literal = "'" + args.batch_id.replace("'", "''") + "'" + query = f"SELECT * FROM {args.layer}.airports" + if args.layer == "stg": + query += f" WHERE _load_id = {batch_literal}" + actual = read_rows("greenplum", query) + errors = compare_snapshot(expected, actual, args.batch_id, args.layer) + if errors: + print("Сверка не прошла. Проверьте SQL загрузки и выбранный Run ID:") + for error in errors[:10]: + print(f"- {error}") + return 1 + print( + f"Сверка пройдена: {args.layer}.airports, _load_id={args.batch_id}, " + f"{len(actual)} ключей; все бизнес-поля и метки совпадают с требованиями." + ) + return 0 + except subprocess.CalledProcessError as error: + print("Не удалось прочитать БД. Проверьте docker compose ps и настройки .env.") + print(error.stderr.strip()) + return 1 + except FileExistsError: + print( + "Контрольный файл уже существует. Сохраните его; для нового опыта выберите другой путь." + ) + return 1 + except (OSError, ValueError, KeyError, TypeError) as error: + print( + f"Не удалось выполнить сверку: {error}. Проверьте путь и формат контрольного файла." + ) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_airports_snapshot_check.py b/tests/test_airports_snapshot_check.py new file mode 100644 index 0000000..e05fc2f --- /dev/null +++ b/tests/test_airports_snapshot_check.py @@ -0,0 +1,80 @@ +"""Проверка должна отвергать ошибки, которые не видны по количеству строк.""" + +import importlib.util +from copy import deepcopy +from pathlib import Path + +import pytest + +spec = importlib.util.spec_from_file_location( + "check_airports_snapshot", + Path(__file__).resolve().parents[1] / "scripts/check_airports_snapshot.py", +) +checker = importlib.util.module_from_spec(spec) +spec.loader.exec_module(checker) + +SOURCE = [ + { + "airport_code": "AAA", + "airport_name": '{"en": "Airport", "ru": "Аэропорт"}', + "city": '{"ru": "Город"}', + "country": '{"ru": "Страна"}', + "coordinates": "(1,2)", + "timezone": "Europe/Moscow", + } +] + + +def snapshot(): + """Снимок с одним ключом и обязательными метками загрузки.""" + rows = deepcopy(SOURCE) + rows[0].update( + _load_id="B", _load_ts="2026-09-22 12:00:00", event_ts="2026-09-22 12:00:00" + ) + return rows + + +def test_accepts_complete_snapshot_with_json_key_order_difference(): + rows = snapshot() + rows[0]["airport_name"] = '{"ru":"Аэропорт","en":"Airport"}' + assert checker.compare_snapshot(SOURCE, rows, "B", "stg") == [] + + +def test_accepts_different_row_order_from_segments(): + """Порядок выдачи строк Greenplum не меняет содержимое снимка.""" + expected = deepcopy(SOURCE) + deepcopy(SOURCE) + expected[1]["airport_code"] = "BBB" + rows = snapshot() + snapshot() + rows[1]["airport_code"] = "BBB" + assert checker.compare_snapshot(expected, rows[::-1], "B", "stg") == [] + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("airport_code", "BBB"), + ("airport_name", '{"ru":"Старое имя","en":"Airport"}'), + ("country", '{"ru":"Другая страна"}'), + ("coordinates", "(2,1)"), + ("_load_id", "A"), + ("event_ts", None), + ], +) +def test_rejects_wrong_values_despite_equal_row_count(field, value): + rows = snapshot() + rows[0][field] = value + assert checker.compare_snapshot(SOURCE, rows, "B", "stg") + + +def test_rejects_duplicate_and_empty_snapshot(): + rows = snapshot() + assert checker.compare_snapshot(SOURCE, rows + rows, "B", "stg") + assert checker.compare_snapshot(SOURCE, [], "B", "stg") + + +def test_ods_requires_both_transformation_and_selected_batch(): + rows = snapshot() + rows[0].update(airport_name="Аэропорт", city="Город", country="Страна") + assert checker.compare_snapshot(SOURCE, rows, "B", "ods") == [] + rows[0]["_load_id"] = "A" + assert checker.compare_snapshot(SOURCE, rows, "B", "ods") diff --git a/tests/test_dags_smoke.py b/tests/test_dags_smoke.py index cfe2555..45c216c 100644 --- a/tests/test_dags_smoke.py +++ b/tests/test_dags_smoke.py @@ -66,6 +66,22 @@ def test_bookings_stg_ddl_dag_structure(): _assert_reachable(dag, "apply_stg_bookings_ddl", task_id) +def test_lab_pxf_airports_uses_snapshot_sql_and_reference_dependencies(): + """Опыт читает четыре справочника; генератор и транзакции в него не входят.""" + dag = _load_dag("airflow.dags.lab_pxf_airports") + assert len(dag.tasks) == 8 + for entity in ("airports", "airplanes", "routes", "seats"): + load = dag.get_task(f"load_{entity}_to_stg") + dq = dag.get_task(f"check_{entity}_dq") + assert load.sql == f"stg/{entity}_load.sql" + assert dq.sql == f"stg/{entity}_dq.sql" + assert load.postgres_conn_id == dq.postgres_conn_id == "greenplum_conn" + _assert_direct_edge(dag, load.task_id, dq.task_id) + _assert_direct_edge(dag, "check_airports_dq", "load_routes_to_stg") + _assert_direct_edge(dag, "check_airplanes_dq", "load_routes_to_stg") + _assert_direct_edge(dag, "check_airplanes_dq", "load_seats_to_stg") + + def test_bookings_to_gp_stage_dag_structure(): """Проверка структуры DAG bookings_to_gp_stage.""" dag = _load_dag("airflow.dags.bookings_to_gp_stage")