316 lines
17 KiB
Markdown
316 lines
17 KiB
Markdown
# Проверка лабораторной 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
|
||
```
|