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 b21d7a6..ede985c 100644 --- a/README.md +++ b/README.md @@ -112,6 +112,7 @@ SELECT COUNT(*) FROM dm.route_performance; - `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/design/pxf_airports_lab.md), без генерации транзакций. - `bookings_ods_ddl` — создаёт/обновляет ODS-таблицы по домену bookings. - `bookings_to_gp_ods` — загружает данные из STG в ODS (SCD1 UPSERT) и выполняет 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/design/labs_development.md b/docs/design/labs_development.md index e0d1700..cd9374d 100644 --- a/docs/design/labs_development.md +++ b/docs/design/labs_development.md @@ -59,7 +59,7 @@ ## Принято: прием справочника входит в обязательную программу -Добавить лабораторную **«Восстановите прием справочника из источника»**. +Добавлена лабораторная **«Восстановите прием справочника из источника»**. Она закрывает практику подключения через PXF и загрузки полного снимка в STG, которые сейчас даны готовыми. @@ -69,11 +69,13 @@ в том же батче и получение нового снимка. Результат должен снова участвовать в действующей цепочке ODS → DDS → DM. -Конкретный справочник, состав удаляемых файлов, место в маршруте и проверки -еще предстоит выбрать. Кандидат — `airports`: рядом есть -[внешняя и внутренняя таблицы](https://git.dementev.space/ddmitry/airflow-greenplum/src/branch/main/sql/stg/airports_ddl.sql), -[загрузка снимка](https://git.dementev.space/ddmitry/airflow-greenplum/src/branch/main/sql/stg/airports_load.sql) и -[преобразование в ODS](https://git.dementev.space/ddmitry/airflow-greenplum/src/branch/main/sql/ods/airports_load.sql). +В [#9](https://git.dementev.space/ddmitry/airflow-greenplum/issues/9) выбран `airports`; +место — после быстрого старта и знакомства с моделью, перед снимками ODS. +Студент восстанавливает внешний DDL и загрузку, сохраняя готовыми внутреннюю +таблицу, JDBC-конфигурацию, DQ и DAG. Перед реализацией очищает старые снимки +только `stg.airports`. Опыты с повторным выполнением, изменением источника и +восстановлением описаны в [инструкции main](https://git.dementev.space/ddmitry/airflow-greenplum/src/branch/main/docs/assignment/pxf_airports.md); +обоснование и эталон — в [материалах автора](pxf_airports_lab.md). ## Покрытие обязательного цикла @@ -82,7 +84,7 @@ | Участок | Что студент делает сейчас | Что стоит добавить | |---|---|---| -| PXF и снимок STG | Запускает готовую цепочку | Принятая лабораторная по приему справочника | +| PXF и снимок STG | Восстанавливает прием `airports`, испытывает повтор, изменение источника и передачу в ODS | Закрыто лабораторной #9 | | Инкремент STG | Использует готовый отбор по дате источника | Самостоятельную загрузку новых данных и проверку границы отбора | | Снимки ODS | Реализует `airplanes`, `seats` и их DQ | Сохранить задания; использовать их после знакомства с приемом данных | | Инкремент ODS | Использует готовые загрузки | Обработку новых ключей и изменений уже полученных строк | @@ -137,7 +139,7 @@ | Новая работа | Состояние перед началом | Результат для продолжения | |---|---|---| -| Прием справочника | Пройден быстрый старт, работает эталонная цепочка до `sales_report`; выбран участок для вычитания | Собственная загрузка получает новый снимок STG, который участвует в общем батче справочников и принимается ODS | +| Прием справочника | Пройден быстрый старт, работает эталонная цепочка до `sales_report` | Собственная загрузка проверена; источник восстановлен, новый общий снимок принят ODS и готов для заданий `airplanes` и `seats` | | Инкремент STG/ODS | Разобраны прием снимка и существующие задания ODS; подготовлены таблицы и входные пачки | Собственные загрузки дают ожидаемое текущее состояние ODS после всех проверочных случаев | | Сборка факта | Реализованы и проверены измерения DDS | Факт построен собственным SQL, проверены ключи и значения; на нем можно выполнять задания DM | | Оркестрация и восстановление | SQL выбранного участка реализован и проверен | DAG выполняет загрузку и DQ в нужном порядке; после подготовленного сбоя обработка завершается с ожидаемыми данными | @@ -147,8 +149,9 @@ не может быть условием начала первой работы. Действующие особенности запусков и состояния слоев описаны в [порядке выполнения DAG](https://git.dementev.space/ddmitry/airflow-greenplum/src/branch/main/docs/dag_execution_order.md). -Предложения основаны на чтении исходников; воспроизводимость новых -сценариев на стенде еще не проверялась. +Оставшиеся предложения основаны на чтении исходников; их воспроизводимость +на стенде еще не проверялась. Состояние проверки приема `airports` +зафиксировано отдельно в [материалах #9](pxf_airports_lab.md#проверочный-прогон). ## Как подготовить лабораторную diff --git a/docs/design/pxf_airports_lab.md b/docs/design/pxf_airports_lab.md new file mode 100644 index 0000000..9256a18 --- /dev/null +++ b/docs/design/pxf_airports_lab.md @@ -0,0 +1,128 @@ +# Прием airports через PXF: материалы автора + +Лабораторная [#9](https://git.dementev.space/ddmitry/airflow-greenplum/issues/9) +из [серии #8](https://git.dementev.space/ddmitry/airflow-greenplum/issues/8). +Студент работает по [инструкции в main](https://git.dementev.space/ddmitry/airflow-greenplum/src/branch/main/docs/assignment/pxf_airports.md). +Общие принципы программы находятся в [авторской заметке](labs_development.md). + +## Место и самостоятельная часть + +Работа идет после быстрого старта и знакомства с моделью, перед существующими +`ods.airplanes` и `ods.seats`. Входом служит прошедшая эталонная цепочка +до `dm.sales_report` на учебной `main`; пустые студенческие таблицы не мешают +начать работу. Полный `bookings_validate` не является условием входа. + +Выбран `airports`: он небольшой, содержит ключ, многоязычные JSON и координаты, +а его готовая загрузка ODS дает наблюдаемый результат преобразования. +Студент удаляет внешний DDL из `sql/stg/airports_ddl.sql` и заменяет +`sql/stg/airports_load.sql` заглушкой. В базе удаляет внешний объект и все +снимки только `stg.airports`. Без очистки прошлые строки могли бы скрыть +отсутствие собственной загрузки. + +Внутренний DDL, JDBC-конфигурация и DQ остаются готовыми. Добавлен готовый +`lab_pxf_airports`: восемь PostgresOperator выполняют существующие загрузки +и проверки четырех справочников. Студент задает +состав и порядок полей внешней таблицы, `LOCATION`, затем пишет получение +снимка с ключом и метками. Близкий пример — `airplanes`. Физическая организация +внутренней таблицы и самостоятельная оркестрация относятся к следующим работам. + +## Устройство урока + +Основной текст строится вокруг собственной загрузки и опытов с данными. +Сначала студент пишет SQL по требованиям и близкому примеру, затем +предполагает результат повторного выполнения и переименования аэропорта, +наблюдает данные и объясняет различия. Передача нового названия в ODS +показывает применение его работы в следующем слое. + +Команды подготовки, первые действия Airflow, сверка и восстановление +находятся в раскрываемых блоках рядом с соответствующим действием. +Проверки остаются готовыми: написание собственной DQ относится к дальнейшим +заданиям. Результат студента — два SQL-файла и объяснение опытов; собирать +отчет из JSON, Run ID и логов не требуется. + +Обозначения A/B/C, проверки намеренно неверной реализации и полный протокол +приемки относятся к авторским материалам. Они подтверждают пригодность +лабораторной, но не задают структуру урока. Одного сокращения формулировок +было бы недостаточно: оно сохранило бы последовательность сбора доказательств +вместо работы с собственной загрузкой. Демонстрационный разбор также +не дал бы предусмотренной программой самостоятельной практики. + +## Полное решение + +Эталон уже содержится в этой ветке: + +- [airports_ddl.sql](../../sql/stg/airports_ddl.sql): шесть TEXT-полей, + `pxf://bookings.airports_data?PROFILE=JDBC&SERVER=bookings-db`, + `FORMAT 'CUSTOM' (formatter='pxfwritable_import')`. +- [airports_load.sql](../../sql/stg/airports_load.sql): чтение внешней таблицы, + `_load_id = '{{ run_id }}'`, две метки `now()`, защита `NOT EXISTS` + по паре `(_load_id, airport_code)`. +- [airports_dq.sql](../../sql/stg/airports_dq.sql): базовая обратная связь DAG. +- [airports_load.sql для ODS](../../sql/ods/airports_load.sql): выбор батча, + извлечение `ru` и замена текущего снимка. + +Для воспроизведения автор может после вычитания вернуть два исходных файла +из сохраненной копии. В учебной ветке они также остаются готовыми при первом +запуске: вычитание выполняется самим студентом во время работы. +Скрывать ответ технически не требуется; самостоятельность проверяется +опытами и объяснением результата. + +## Почему нужны эти проверки + +Готовая [сверка в main](https://git.dementev.space/ddmitry/airflow-greenplum/src/branch/main/scripts/check_airports_snapshot.py) +экспортирует источник напрямую из Postgres, без внешней таблицы студента. +Поэтому неверный `LOCATION` не становится одновременно ошибкой загрузки +и ее контрольного источника. В JSON сохраняются все бизнес-поля, а не только +выбранное русское имя. Перезапись контрольного файла запрещена. + +В STG проверяются ключи, отсутствие дублей внутри батча, все бизнес-поля, +`_load_id`, наличие `_load_ts` и `event_ts`. JSON сравнивается по содержимому, +чтобы порядок свойств не влиял на результат. В ODS проверяется вся таблица, +русские значения названий и точный STG `_load_id` у каждой строки. +Штатная DQ сверяет количество и обязательные поля, поэтому сама по себе +не отвергнет старое непустое название с правильным числом строк. + +| Случай | Наблюдаемый результат | +|---|---| +| Вычитание | STG пуст; внешняя таблица отсутствует; независимая сверка отказывает | +| A | Все ключи и бизнес-поля совпадают с прямым экспортом источника | +| Повтор A | Clear только загрузки аэропортов, затем ее DQ в том же запуске; строки сохраняются, генератор не повторяется | +| Изменение источника | Сохранен исходный JSON одного аэропорта; изменен только `ru` | +| Неверная старая копия | A или его копия с новым `_load_id` не совпадает с контрольным источником B при равном числе строк | +| B | B содержит новое имя; A все еще совпадает с исходным экспортом | +| ODS | Все четыре справочника имеют B; resolver без `conf` выбирает B; ODS содержит B и новое имя | +| Восстановление C | Источник в точности равен исходному A; новый общий C проходит ODS → DDS → DM | + +Для повторов A используется Clear отдельных задач. A/B/C создаются +через `lab_pxf_airports`: повтор снимка не требует нового транзакционного дня. +Это стало отдельной границей при подготовке опыта: полные запуски STG +привязывали каждый контроль справочника к длительной генерации Bookings. +Учебный DAG явно исключает генератор и транзакционные загрузки, но использует +те же SQL-файлы снимков. Основной `bookings_to_gp_stage` сохраняет свое поведение. + +Автоматический resolver выбирает последний общий ID по времени готовности. +Наличие ID во всех таблицах не доказывает их полноту — нужны DQ и сверка. +Передача `stg_batch_id` через `conf` обошла бы автоматический выбор, поэтому +в проверочном опыте ODS запускается с пустым `conf`. + +## Выход и переход + +В `stg.airports` остаются A/B/C; старые снимки аэропортов удалены в начале. +В остальных справочниках остаются также снимки быстрого старта. Транзакционные +таблицы сохраняют данные подготовки. B сохраняет +учебное изменение как историю; последний общий C и текущий ODS содержат +восстановленные значения. Источник восстанавливается из постоянной таблицы +`public.lab9_airport_original`, поэтому закрытие psql не теряет копию. +Повтор UPDATE восстановления безопасен; таблица удаляется только после сверки. + +Следующая работа берет общий C для `ods.airplanes` и `ods.seats`, без нового +STG. Затем сохраняется прежний маршрут измерений и витрин. Работы #10–#14 +остаются отдельными задачами серии; эта лабораторная не добавляет их заглушки +и не меняет текущие требования к ним. + +## Проверочный прогон + +Команды, версии и наблюдения сохраняются в +[протоколе прогона](../reports/2026-09-22-pxf-airports.md). +Проверять нужно учебную `main`: полный `solution` заполняет дополнительные +таблицы и не подтверждает проходимость первой работы с учебными заглушками. diff --git a/docs/reports/2026-09-22-pxf-airports.md b/docs/reports/2026-09-22-pxf-airports.md new file mode 100644 index 0000000..37c8510 --- /dev/null +++ b/docs/reports/2026-09-22-pxf-airports.md @@ -0,0 +1,315 @@ +# Проверка лабораторной PXF → STG, 22 сентября 2026 + +Задача [#9](https://git.dementev.space/ddmitry/airflow-greenplum/issues/9) +из [серии #8](https://git.dementev.space/ddmitry/airflow-greenplum/issues/8). +[Материалы автора](../design/pxf_airports_lab.md) описывают границу +самостоятельной работы и эталонное решение. + +## Среда и начальное состояние + +Проверена учебная ветка на базе `main` (`1dd5838`) с добавленными +`lab_pxf_airports` и независимой сверкой. Студенческие заглушки ODS/DDS/DM +сохранены. Полная реализация из `solution` в этот стенд не переносилась. +Версии: Airflow 2.9.2, Greenplum 6.27.1, Postgres 16. + +```bash +make up +make bookings-init +make ddl-gp +``` + +После подготовки выполнена обычная цепочка STG → ODS → DDS → DM с Run ID +`lab9-baseline-stage`, `lab9-baseline-ods`, `lab9-baseline-dds`, +`lab9-baseline-dm`. Все четыре запуска завершились успешно до вычитания. + +В локальном `.env` использованы свободные порты: Airflow 18090, метабаза +15433, Bookings 15434, Greenplum 15435. Airflow ограничен одним web-worker, +`PARALLELISM=2`, `MAX_ACTIVE_TASKS_PER_DAG=2`, `PARSING_PROCESSES=1`. +Эти настройки не внесены в репозиторий; SQL не изменен. + +Первый прогон базового генератора прервался: Postgres сообщил о выходе +server process с кодом 2 и переинициализации, после чего источник оказался +пустым. `OOMKilled=false`, рестарта контейнера не было; доказанной причины +нет. После ограничения ресурсов только своего Airflow источник восстановлен +через `make bookings-init`, а базовая цепочка успешно повторена. +Чужие контейнеры не изменялись. Дальнейшие снимки справочников получены +через `lab_pxf_airports`, без генерации транзакционных дней. + +DAG запускались через `POST /api/v1/dags/{dag_id}/dagRuns` с телом +`{"dag_run_id": "указанный в протоколе Run ID", "conf": {}}`. +Перед следующим шагом проверялось состояние `success` всего запуска. + +## Вычитание, A и повтор + +A/B/C — обозначения авторского прогона. В ученической инструкции это +первая загрузка, новый снимок после переименования и снимок после восстановления. +Файлам `source-A.json` и `source-B.json` в задании соответствуют +`airports-before.json` и `airports-after.json`. + +```bash +mkdir -p data/lab9 +uv run --no-project scripts/check_airports_snapshot.py export data/lab9/source-A.json +``` + +Прямой экспорт Postgres: 5501 аэропорт, все шесть бизнес-полей. +Внешний DDL удален из `sql/stg/airports_ddl.sql`, загрузка заменена +на `SELECT 1;`. В Greenplum: + +```sql +DROP EXTERNAL TABLE IF EXISTS stg.airports_ext; +TRUNCATE TABLE stg.airports; +SELECT COUNT(*) FROM stg.airports; +``` + +Получено 0 строк. Заглушка успешно выполнилась, но сверка отказала: + +```bash +uv run --no-project scripts/check_airports_snapshot.py check lab9-empty data/lab9/source-A.json +``` + +Код 1: `Не получены ключи`, всего 5501. Запрос к `stg.airports_ext` +получил `relation does not exist`. Старые данные не маскируют вычитание. +После возврата эталонных DDL и загрузки применен файл: + +```sql +\set ON_ERROR_STOP on +\i /sql/stg/airports_ddl.sql +SELECT * FROM stg.airports_ext ORDER BY airport_code LIMIT 3; +``` + +PXF вернул AAA, AAC, AAE с исходными JSON, координатами и часовыми поясами. +Успешно выполнен `lab_pxf_airports` с Run ID `lab9-A-stage`. + +```bash +uv run --no-project scripts/check_airports_snapshot.py check lab9-A-stage data/lab9/source-A.json +``` + +Код 0: 5501 ключ, все бизнес-поля и метки соответствуют требованиям. +Повторены только `load_airports_to_stg`, затем `check_airports_dq` +в том же запуске через +`POST /api/v1/dags/lab_pxf_airports/clearTaskInstances`: + +```json +{ + "dag_run_id": "lab9-A-stage", + "task_ids": ["load_airports_to_stg"], + "dry_run": true, + "only_failed": false, + "only_running": false, + "include_upstream": false, + "include_downstream": false, + "include_subdags": false, + "include_parentdag": false, + "reset_dag_runs": true +} +``` + +Предварительный ответ содержал ровно одну выбранную задачу, затем +отправлено то же тело с `dry_run=false`. Для DQ заменен `task_ids`. +Повторная сверка A — код 0. При сравнении по ключу все строки, включая +метки, совпали до и после повтора. Базовый генератор не повторялся. + +В браузере проверены граф, Run ID и окно Clear task. Подтверждение повтора +выполнялось через API. Первоначальный тестовый сценарий ошибочно сравнивал +порядок строк MPP-ответа. После сравнения по ключу повтор подтвердился; +ошибка не относилась к загрузке. Независимая сверка изначально сопоставляет +строки по ключу, экспорт дополнительно сортирует их для файлового сравнения. + +## Изменение источника и неверная копия + +В Postgres выполнено: + +```sql +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; +``` + +Изменен один ключ AAA: `Анаа` → `Анаа [lab9]`, английское `Anaa` +и структура JSON сохранены. Внешняя таблица сразу вернула новое имя, +первый снимок сохранил старое. + +```bash +uv run --no-project scripts/check_airports_snapshot.py export data/lab9/source-B.json +uv run --no-project scripts/check_airports_snapshot.py check lab9-A-stage data/lab9/source-B.json +``` + +В источнике по-прежнему 5501 строка. Проверка — код 1, +`AAA: отличается airport_name`. Затем в Greenplum создана старая копия +с новым ID и заполненными метками: + +```sql +INSERT INTO stg.airports +SELECT airport_code, airport_name, city, country, coordinates, timezone, + now(), now(), 'lab9-bad-copy' +FROM stg.airports WHERE _load_id = 'lab9-A-stage'; +``` + +```bash +uv run --no-project scripts/check_airports_snapshot.py check lab9-bad-copy data/lab9/source-B.json +``` + +Вставлено 5501 строк; проверка отвергла копию с кодом 1 и ошибкой имени. +Контрольные строки удалены: + +```sql +DELETE FROM stg.airports WHERE _load_id = 'lab9-bad-copy'; +``` + +## Снимок B и передача в ODS + +После успешного `lab_pxf_airports`, Run ID `lab9-B-stage`: + +```bash +uv run --no-project scripts/check_airports_snapshot.py check lab9-B-stage data/lab9/source-B.json +uv run --no-project scripts/check_airports_snapshot.py check lab9-A-stage data/lab9/source-A.json +``` + +Обе сверки прошли. Общий батч проверен запросом: + +```sql +SELECT 'airports' AS object, _load_id, COUNT(*) FROM stg.airports WHERE _load_id = 'lab9-B-stage' GROUP BY _load_id +UNION ALL +SELECT 'airplanes', _load_id, COUNT(*) FROM stg.airplanes WHERE _load_id = 'lab9-B-stage' GROUP BY _load_id +UNION ALL +SELECT 'routes', _load_id, COUNT(*) FROM stg.routes WHERE _load_id = 'lab9-B-stage' GROUP BY _load_id +UNION ALL +SELECT 'seats', _load_id, COUNT(*) FROM stg.seats WHERE _load_id = 'lab9-B-stage' GROUP BY _load_id; +``` + +| Таблица | Строк с `lab9-B-stage` | +|---|---:| +| airports | 5501 | +| airplanes | 10 | +| routes | 860 | +| seats | 1741 | + +До запуска ODS команда ниже отказала из-за прежнего батча и названия, +после запуска — прошла: + +```bash +uv run --no-project scripts/check_airports_snapshot.py check lab9-B-stage data/lab9/source-B.json --layer ods +``` + +Выполнен обычный `bookings_to_gp_ods`, Run ID `lab9-B-ods`, `conf={}`. +XCom `return_value` задачи `resolve_stg_batch_id` равен `lab9-B-stage`. +Все 5501 строк ODS совпали с ожидаемыми данными и меткой B; +название AAA — `Анаа [lab9]`. Выбор через `conf` не подменял resolver. + +## Восстановление и продолжение + +В Postgres дважды выполнен UPDATE восстановления: + +```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; +``` + +Оба раза `UPDATE 1`, исходный JSON восстановлен. Файловая проверка: + +```bash +uv run --no-project scripts/check_airports_snapshot.py export data/lab9/source-C.json +cmp data/lab9/source-A.json data/lab9/source-C.json +``` + +Файлы совпали побайтово. Затем успешно выполнены: + +| DAG | Run ID | +|---|---| +| lab_pxf_airports | lab9-C-stage | +| bookings_to_gp_ods | lab9-C-ods | +| bookings_to_gp_dds | lab9-C-dds | +| bookings_to_gp_dm | lab9-C-dm | + +Resolver с пустым `conf` выбрал `lab9-C-stage`. Сверки A, B и C со своими +источниками, а также всей ODS с C прошли. + +```sql +SELECT _load_id, COUNT(*) FROM stg.airports GROUP BY _load_id ORDER BY _load_id; +SELECT airport_bk, airport_name FROM dds.dim_airports ORDER BY airport_bk LIMIT 1; +SELECT COUNT(*) FROM dds.fact_flight_sales; +SELECT COUNT(*) FROM dm.sales_report; +``` + +В STG остались A/B/C по 5501 строк каждый. В `dds.dim_airports` у AAA +снова `Анаа`. Факт содержит 1 608 696 строк, `dm.sales_report` — 25 732. +Остальные справочники сохраняют также батчи быстрого старта; +транзакционные данные не менялись во время A/B/C. Новый общий снимок +пригоден для заданий `ods.airplanes` и `ods.seats`. +Полный `bookings_validate` не запускался: будущие задания остаются заглушками. + +## Проверки кода и документации + +- `make lint` — успешно. +- `make test` — 14 passed, 15 skipped: десять тестов сверки и четыре + SQL-контракта прошли. Четырнадцать smoke-тестов DAG пропущены из-за + существующих заглушек Airflow в `conftest.py`; один тест ODS требует + явного включения интеграционного прогона. +- В настоящем Airflow отдельно проверен DagBag: ошибок импорта нет, + восемь задач нового DAG, SQL-шаблоны раскрываются в исходные файлы, + зависимости загрузок и DQ верны. REST API не сообщил ошибок импорта. +- Проверены локальные ссылки и актуальность карты. В браузере пройден + маршрут README → задания → PXF → ODS и ссылка карточки `stg.airports`. +- `git diff --check` пройден в учебной и авторской рабочих копиях. + +Учебный прогон выполнен до редакционной перестройки урока. Сценарии, +SQL загрузки, DAG и сверка сохранены; изменились порядок объяснений, +имена файлов в примерах и показ подсказок. После редактирования проверены +296 локальных ссылок учебной ветки и 9 ссылок авторских материалов, +синтаксис bash-примеров, актуальность карты (39 объектов, 90 связей, +216 ссылок). В локальном HTML-предпросмотре Chromium раскрыты все восемь +подсказок, проверено отображение девяти блоков кода и переход к ODS. + +Упрощенное завершение урока использует исходный контроль вместо третьего +файла. На сохраненном состоянии прогона отдельно выполнены обе команды: + +```bash +uv run --no-project scripts/check_airports_snapshot.py check lab9-C-stage data/lab9/source-A.json +uv run --no-project scripts/check_airports_snapshot.py check lab9-C-stage data/lab9/source-A.json --layer ods +``` + +Обе прошли: 5501 ключ, все бизнес-поля и метки соответствуют требованиям. +Полный прогон после редакционной правки не повторялся. + +## Проверка исправления после ревью + +Независимое ревью Standards не обнаружило нарушений. Ревью Spec нашло +ловушку при исправлении SQL: уже записанные дубли остаются в батче после +добавления `NOT EXISTS`. Инструкция дополнена очисткой только ошибочного +`_load_id`, восстановлением снимка и повтором на заполненном снимке. +Исправление повторно просмотрено; замечание закрыто. + +В отдельном батче `lab9-review-retry` дважды выполнен INSERT всех строк +`stg.airports_ext` с текущими метками. Получено 11002 строки, сверка отказала +из-за дублей. После выполнения эталонного `airports_load.sql` с этим же +Run ID осталось 11002 строки; сверка снова отказала. + +```sql +DELETE FROM stg.airports WHERE _load_id = 'lab9-review-retry'; +``` + +После очистки эталонная загрузка дала 5501 строку и успешную сверку. +Ее повтор также прошел. Временный батч затем удален тем же DELETE; +состав и количества A/B/C сохранились. Это отдельный SQL-опыт восстановления, +а не повтор всего учебного маршрута. + +Перед публикацией `make lint` и `make test` снова прошли +(14 passed, 15 skipped). Состояния семи запусков A/B/C и нижних слоев +перечитаны через Airflow API: все `success`. Выдержка настоящего лога +`check_airports_dq` первого запуска: + +```text +[2026-09-22T17:42:17.505+0300] {taskinstance.py:1206} INFO - Marking task as SUCCESS. dag_id=lab_pxf_airports, task_id=check_airports_dq, run_id=lab9-A-stage, execution_date=20260922T144212, start_date=20260922T144216, end_date=20260922T144217 +[2026-09-22T17:42:17.586+0300] {local_task_job_runner.py:240} INFO - Task exited with return code 0 +``` 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 f88c68c..d92dfae 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")