ddmitry/airflow-greenplum
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs(plans): добавлен план валидационного DAG, архивирован routes-to-reference

Browse Source 
- Зачем:
  - план routes-to-reference выполнен, нужен план следующего этапа — валидационный DAG
- Что:
  - перемещён docs/plans/2026-03-11_routes-to-reference.md → docs/archive/
  - создан docs/plans/2026-03-12_validation-dag.md: структура DAG (17 SQL + DAG + smoke-тест),
    активный тест SCD2 (цепочка тасков с безопасным откатом), прошёл 5 раундов ревью ChatGPT
- Проверка:
  - make test

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
ddadmin
2026-03-12 11:50:03 +03:00
co-authored by Claude Opus 4.6
parent 4e9d8f5a1c
commit 23cde47fe7
2 changed files with 999 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/plans/2026-03-11_routes-to-reference.md
-495
View File
@@ -1,495 +0,0 @@
# Plan: Перенос STG в эталон, ODS airplanes+seats — студенту
> Версия: 4 (после третьего ревью ChatGPT 5.4)
> Дата: 2026-03-11
## Контекст и мотивация
### Обнаруженная проблема
При подготовке к Этапу 4 (раскладка по веткам main / solution) обнаружен **конфликт
между текущим дизайном БД** (`docs/design/db_schema.md`, `docs/design/bookings_dds_design.md`)
**и планом курсового задания** (`docs/design/assignment_design.md`).
Суть конфликта: эталонный пайплайн (который должен работать «из коробки» на main-ветке
для студента) **неработоспособен**, если студент ещё не реализовал ни одного задания.
Три DDS-измерения (`dim_routes`, `dim_passengers`, `dim_airplanes`) отнесены к заданию
студента, но эталонная фактовая таблица `fact_flight_sales` зависит от них через
цепочку LEFT JOIN. Без этих измерений факт загружается с массовыми NULL в ключевых FK,
а DQ-проверки блокируют pipeline. Эталонные витрины (`sales_report`, `airport_traffic`)
становятся бесполезными.
Проблема усугубляется архитектурой batch resolver ODS DAG и тестовыми контрактами,
которые требуют данных во **всех** snapshot-таблицах STG.
Ниже — детальный разбор и выбранное решение.
### Фактовая таблица: зависимость от студенческих измерений
`fact_flight_sales` зависит от **всех 7 измерений** DDS,
включая 3 студенческих: `dim_routes`, `dim_passengers`, `dim_airplanes`.
Критическая цепочка: аэропорты в факте разрешаются **через `dim_routes`**:
```
flt.route_no → dim_routes.route_bk → rte.departure_airport → dim_airports.airport_bk → airport_sk
```
Если `dim_routes` пуст (студент ещё не реализовал), то **NULL** получают не только
`route_sk`, но и оба `airport_sk` и `airplane_sk`. Эталонная витрина `sales_report`
становится бесполезной.
Причина: `ods.flights` содержит только `route_no`, но не `departure_airport` /
`arrival_airport` / `aircraft_code`. Эти поля доступны только через таблицу `routes`.
### Почему нельзя просто использовать DDL-заглушки
Если `dim_routes` — пустая заглушка, все LEFT JOIN через неё возвращают NULL.
Факт загружается, но:
- `departure_airport_sk = NULL` → `sales_report` не может группировать по аэропортам
- `arrival_airport_sk = NULL` → `airport_traffic` не может считать трафик
- UPDATE факта **не перезаписывает SK** (by design, строка 4-5 load-скрипта) →
NULL-и остаются навсегда до TRUNCATE + перезагрузки
### Дополнительные архитектурные ограничения (из ревью)
1. **Fact DQ блокирует загрузку** (`fact_flight_sales_dq.sql`):
- Строка 71: `passenger_sk IS NULL` → **hard fail** (100% NULL → exception)
- Строка 98: `route_sk / airplane_sk IS NULL` → при >1% строк → **exception**
- На main с пустыми студенческими dims — 100% NULL → pipeline падает
2. **Batch resolver ODS DAG** (`bookings_to_gp_ods.py:59-77`):
- `INTERSECT` по `stg.airports`, `stg.airplanes`, `stg.routes`, `stg.seats`
- Если **любая** STG-таблица пуста — общего batch нет → весь ODS pipeline падает
- Контракт зафиксирован тестом `test_ods_sql_contract.py:30`
### Выбранное решение
**Весь STG-слой — эталонный** (все 9 таблиц загружаются reference-кодом).
Это гарантирует:
- Batch resolver всегда находит согласованный batch (все 4 snapshot-таблицы заполнены)
- DAG-зависимости и smoke-тесты STG не требуют изменений
- Никакого расхождения batch-контракта между main и solution
**ODS: `airplanes` + `seats` остаются заданием студента** (TRUNCATE+INSERT практика).
**DDS: `dim_airplanes`, `dim_passengers`, `dim_routes`** — задание студента (DDL-заглушки).
**Fact lookup**: аэропорты через `ods.routes` (эталон); `airplane_sk` остаётся
point-in-time через `dim_routes` (NULL на main — допустимо).
**Fact DQ**: на main ослабить проверки студенческих SK.
### Что теряет студент (и почему это приемлемо)
| Потеря | Компенсация |
|--------|-------------|
| STG-практика (PXF external tables) | Изучает эталонный STG-код; отдельный PXF-практикум в бэклоге (`assignment_design.md`, раздел 6) |
| `ods.routes` (TRUNCATE+INSERT) | Остаются `ods.airplanes` + `ods.seats` — тот же паттерн |
Ключевые элементы задания **сохранены**:
- ODS: 2 таблицы (airplanes, seats) — практика TRUNCATE+INSERT
- DDS: `dim_airplanes` (SCD1), `dim_passengers` (SCD1), **`dim_routes` (SCD2)** — ключевой вызов
- DM: 4 витрины разной сложности (от простой к сложной)
### Пересчёт факта после реализации студентом
После реализации всех измерений студенту нужно:
1. Запустить загрузку DDS-измерений (dim_airplanes, dim_passengers, dim_routes)
2. `TRUNCATE dds.fact_flight_sales;`
3. Перезапустить загрузку факта → теперь все SK заполнены
4. Перезапустить DM-витрины
Это стандартная практика при late-arriving dimensions и хорошая обучающая точка.
### airport_traffic — студенческое задание
`airport_traffic` архитектурно не зависит от студенческих dims (использует только
`calendar_sk`, `departure_airport_sk`, `arrival_airport_sk`). Но это **не значит**,
что витрина готова: SQL витрины пишет студент. Данные в факте есть (эталон),
студент пишет агрегирующий SQL. Статус: **задание студента** (как и остальные 3 DM).
---
## Матрица зависимостей DM-витрин
| DM витрина | calendar | airports | tariffs | routes | passengers | airplanes | Работает без студ. dims? |
|---|---|---|---|---|---|---|---|
| `sales_report` (эталон) | + | + | + | — | — | — | **ДА** |
| `airport_traffic` (студент) | + | + | — | — | — | — | Данные есть, SQL пишет студент |
| `route_performance` (студент) | + | — | — | **+** | — | — | Нет |
| `monthly_overview` (студент) | + | — | — | **+** | **+** | **+** | Нет |
| `passenger_loyalty` (студент) | + | — | + | **+** | **+** | — | Нет |
---
## Изменения (код)
### 1. `sql/dds/fact_flight_sales_load.sql` — lookup аэропортов через ods.routes
**Ветка:** chore/bookings-etl (сейчас) + main (Этап 4)
**Текущее состояние** (строки 51-62):
```sql
LEFT JOIN dds.dim_routes AS rte
ON rte.route_bk = flt.route_no
AND flt.scheduled_departure::DATE >= rte.valid_from
AND (rte.valid_to IS NULL OR flt.scheduled_departure::DATE < rte.valid_to)
...
LEFT JOIN dds.dim_airports AS dep
ON dep.airport_bk = rte.departure_airport -- через dim_routes!
LEFT JOIN dds.dim_airports AS arr
ON arr.airport_bk = rte.arrival_airport -- через dim_routes!
LEFT JOIN dds.dim_airplanes AS ap
ON ap.airplane_bk = rte.airplane_code -- через dim_routes!
```
**После изменения:**
```sql
-- Учебный комментарий: Airport lookup через ods.routes (эталонный справочник),
-- а не через dds.dim_routes (студенческое задание SCD2).
-- Это архитектурное решение: эталонный пайплайн работает независимо от студенческого кода.
-- ROW_NUMBER по validity DESC: выбираем актуальную версию расписания маршрута.
-- Аэропорты вылета/прилёта одинаковы во всех версиях одного route_no.
LEFT JOIN (
SELECT route_no, departure_airport, arrival_airport
FROM (
SELECT route_no, departure_airport, arrival_airport,
ROW_NUMBER() OVER (PARTITION BY route_no ORDER BY validity DESC) AS rn
FROM ods.routes
) ranked
WHERE rn = 1
) AS ods_rte ON ods_rte.route_no = flt.route_no
LEFT JOIN dds.dim_routes AS rte
ON rte.route_bk = flt.route_no
AND flt.scheduled_departure::DATE >= rte.valid_from
AND (rte.valid_to IS NULL OR flt.scheduled_departure::DATE < rte.valid_to)
...
LEFT JOIN dds.dim_airports AS dep
ON dep.airport_bk = ods_rte.departure_airport -- через ods.routes (эталон)
LEFT JOIN dds.dim_airports AS arr
ON arr.airport_bk = ods_rte.arrival_airport -- через ods.routes (эталон)
LEFT JOIN dds.dim_airplanes AS ap
ON ap.airplane_bk = rte.airplane_code -- через dim_routes (point-in-time, как прежде)
```
**Почему аэропорты через ods.routes, а airplane через dim_routes:**
- **Аэропорты**: `departure_airport` и `arrival_airport` одинаковы во всех версиях
одного `route_no` (маршрут SVO→LED всегда SVO→LED). Безопасно брать из ODS.
Это даёт эталонным витринам (`sales_report`, `airport_traffic`) корректные `airport_sk`
даже без студенческого `dim_routes`.
- **Самолёт**: `airplane_code` теоретически может отличаться между версиями маршрута.
Текущий контракт DDS (`bookings_dds_design.md:499`) фиксирует, что `airplane_sk`
приходит из **той же point-in-time версии** маршрута, что и `route_sk`.
Менять эту семантику — изменение модели данных, а не стабилизация.
На main `airplane_sk` будет NULL (dim_routes — заглушка). Это допустимо:
ни `sales_report`, ни `airport_traffic` не используют `airplane_sk`.
### 2. `sql/dds/fact_flight_sales_dq.sql` — ослабить проверки студенческих SK (только main)
**Ветка:** только main (Этап 4). На solution-ветке полные проверки сохраняются.
Изменения:
- **Строки 65-75** (`passenger_sk IS NULL`): заменить `RAISE EXCEPTION` на `RAISE NOTICE`.
На main `dim_passengers` — заглушка, 100% строк будут с `passenger_sk IS NULL`.
Любой порог (даже >1%) даст exception. **Только логирование, без блокировки.**
- **Строки 89-108** (route-related FK): разделить на два блока:
**Блок A — эталонные FK (проверка с порогом 1%, как calendar_sk):**
```sql
-- departure_airport_sk и arrival_airport_sk заполняются через ods.routes (эталон).
-- NULL здесь — аномалия данных (пропущен маршрут в ODS), а не отсутствие студенческого кода.
-- Порог 1% — защита от единичных аномалий источника (аналогично calendar_sk).
SELECT COUNT(*) INTO v_null_airport
FROM dds.fact_flight_sales
WHERE departure_airport_sk IS NULL OR arrival_airport_sk IS NULL;
IF v_null_airport > 0 THEN
IF v_null_airport * 100.0 / NULLIF(v_row_count, 0) > 1.0 THEN
RAISE EXCEPTION 'DQ FAILED: NULL airport_sk: % (>1%%)', v_null_airport;
ELSE
RAISE NOTICE 'DQ WARNING: NULL airport_sk: % (<=1%%, допустимо)', v_null_airport;
END IF;
END IF;
```
**Блок B — студенческие FK (только логирование, RAISE NOTICE):**
```sql
-- route_sk и airplane_sk будут NULL, пока студент не реализует dim_routes.
-- Не блокируем pipeline.
SELECT COUNT(*) INTO v_null_student
FROM dds.fact_flight_sales
WHERE route_sk IS NULL OR airplane_sk IS NULL;
IF v_null_student > 0 THEN
RAISE NOTICE 'DQ INFO: студенческие SK (route/airplane) NULL: %. '
'После реализации dim_routes: TRUNCATE fact → перезагрузка.',
v_null_student;
END IF;
```
Итоговое правило на main:
- `departure_airport_sk`, `arrival_airport_sk` — **проверка с порогом 1%** (эталон, через ods.routes; >1% → EXCEPTION, <=1% → NOTICE)
- `passenger_sk`, `route_sk`, `airplane_sk` — **только RAISE NOTICE** (студенческие заглушки, 100% NULL допустимо)
```sql
-- Учебный комментарий: route_sk, airplane_sk и passenger_sk будут NULL,
-- пока вы не реализуете соответствующие DDS-измерения.
-- После реализации: TRUNCATE dds.fact_flight_sales → перезагрузка → все SK заполнены.
-- Полную версию DQ (с блокировкой) см. в ветке solution.
```
### 3. ODS Routes DQ: убрать RI-проверку airplane_code (только main)
**Ветка:** только main (Этап 4). На solution-ветке полные проверки сохраняются.
> **Почему только ODS, не STG?** Весь STG — эталон, `stg.airplanes` всегда заполнен,
> `stg/routes_dq.sql` RI-проверка к `stg.airplanes` проходит штатно. Менять STG не нужно.
> На уровне ODS `ods.airplanes` — студенческая заглушка (пустая), поэтому
> `ods/routes_dq.sql` RI-проверка к `ods.airplanes` упадёт.
**Файл:**
- `sql/ods/routes_dq.sql` — удалить блок строк 143-156 (RI к `ods.airplanes`)
Добавить комментарий:
```sql
-- Учебный комментарий: проверка RI airplane_code → ods.airplanes не выполняется,
-- т.к. таблица ods.airplanes реализуется студентом. После реализации — раскомментируйте
-- (см. ветку solution для полной версии).
```
### 4. ODS DAG-зависимость: убрать airplanes → routes (только main)
**Ветка:** только main (Этап 4)
> **Почему только ODS, не STG?** STG DAG не меняется: `stg.airplanes` — эталон,
> барьер `check_airplanes_dq >> load_routes_to_stg` работает штатно.
> На уровне ODS `dq_ods_airplanes` — заглушка (`SELECT 1;`), формально проходит,
> но зависимость вводит в заблуждение. Убираем для ясности.
`airflow/dags/bookings_to_gp_ods.py` (строка 266):
```python
# Было:
[dq_ods_airports, dq_ods_airplanes] >> load_ods_routes
# Стало:
dq_ods_airports >> load_ods_routes
```
### 5. Заглушки для студенческих файлов (только main)
**Ветка:** только main (Этап 4)
#### STG-слой: весь код остаётся (эталон), заглушки не нужны
Все 9 STG-таблиц загружаются эталонным кодом. Batch resolver работает без изменений.
#### ODS-слой: заглушки для airplanes + seats
| Файл | main (студент) | solution |
|------|----------------|----------|
| `sql/ods/airplanes_ddl.sql` | Полный DDL (таблица нужна) | Тот же |
| `sql/ods/airplanes_load.sql` | Заглушка: `SELECT 1; -- TODO` | Полная реализация |
| `sql/ods/airplanes_dq.sql` | Заглушка: `SELECT 1;` | Полная реализация |
| `sql/ods/seats_ddl.sql` | Полный DDL | Тот же |
| `sql/ods/seats_load.sql` | Заглушка | Полная реализация |
| `sql/ods/seats_dq.sql` | Заглушка | Полная реализация |
#### DDS-слой: заглушки для 3 измерений
| Файл | main (студент) | solution |
|------|----------------|----------|
| `sql/dds/dim_routes_ddl.sql` | Полный DDL (нужен для LEFT JOIN) | Тот же |
| `sql/dds/dim_routes_load.sql` | Заглушка: `SELECT 1; -- TODO: реализуйте SCD2` | Полная реализация |
| `sql/dds/dim_routes_dq.sql` | Заглушка: `SELECT 1;` | Полная реализация |
| `sql/dds/dim_passengers_ddl.sql` | Полный DDL | Тот же |
| `sql/dds/dim_passengers_load.sql` | Заглушка | Полная реализация |
| `sql/dds/dim_passengers_dq.sql` | Заглушка | Полная реализация |
| `sql/dds/dim_airplanes_ddl.sql` | Полный DDL | Тот же |
| `sql/dds/dim_airplanes_load.sql` | Заглушка | Полная реализация |
| `sql/dds/dim_airplanes_dq.sql` | Заглушка | Полная реализация |
#### DM-слой: заглушки для 4 витрин
Аналогично: DDL остаётся, load/dq заменяются заглушками.
Файлы: `airport_traffic`, `route_performance`, `monthly_overview`, `passenger_loyalty`.
---
## Изменения (документация)
### 6. `docs/assignment/analyst_spec.md`
- **Удалить** разделы 1.1-1.3 (все STG-задания) — весь STG теперь эталон
- **Удалить** разделы 2.3 (`ods.routes`) — routes теперь эталон
- **Обновить** рекомендуемый порядок: начинается с ODS (airplanes, seats)
- **Добавить** мотивационную секцию «Почему STG уже реализован»:
- Эталонный пайплайн требует согласованного batch по всем snapshot-таблицам
- Без данных в STG невозможна загрузка ODS и далее по цепочке
- Студент изучает эталонный STG-код как образец
- **Добавить** секцию «Пересчёт факта после реализации измерений»:
- Инструкция: TRUNCATE fact + re-run после реализации всех DDS-измерений
- Педагогическая ценность: late-arriving dimensions, dependency management
### 7. `docs/design/assignment_design.md`
- **Обновить** таблицу «Эталонные таблицы»: добавить весь STG + ods.routes
- **Обновить** таблицу «Задание студенту»: убрать STG целиком, ODS оставить airplanes+seats
- **Обновить** рекомендуемый порядок: начинается с ODS
- **Обновить** структуру валидационного DAG (раздел 4): убрать `validate_stg` секцию
### 8. `docs/design/bookings_dds_design.md`
- **Строки 463, 468** (SQL-скелет факта): обновить JOIN-блок —
`dep.airport_bk` и `arr.airport_bk` теперь через `ods_rte`, а не через `rte`;
`ap.airplane_bk` остаётся через `rte` (point-in-time)
- **Строка 499-501** (текстовое пояснение): обновить —
`departure_airport_sk` и `arrival_airport_sk` берутся из `ods.routes` (эталон);
`airplane_sk` остаётся point-in-time через `dim_routes`
- **Строка 558** (passenger_sk = 0): убрать утверждение о нулевой толерантности —
на main `passenger_sk` будет 100% NULL (dim_passengers — заглушка)
- **Строка 559** (DQ текстовое описание): обновить — на main student SK
(`passenger_sk`, `route_sk`, `airplane_sk`) не блокируют pipeline;
`airport_sk` проверяются с порогом 1%
- **Строка 741** (DQ SQL-пример): обновить встроенный SQL — разделить route-related
блок на эталонный (airport_sk, порог 1%) и студенческий (route_sk, airplane_sk, NOTICE)
### 9. `docs/design/db_schema.md`
- Пометить весь STG и `ods.routes` как эталонные (не студенческие)
- Строка 3: «Все слои реализованы» — уточнить, что на main ODS airplanes/seats,
DDS student dims и DM student vitrines — заглушки
- Строка 253: DQ-контракт «SQL-скрипты с RAISE EXCEPTION» — добавить оговорку,
что на main студенческие DQ-скрипты являются заглушками
### 10. Тесты
- `tests/test_dags_smoke.py` строки 242-243:
На main убрать assert барьера `dq_ods_airplanes → dq_ods_routes` (ODS routes
не зависит от ODS airplanes на main). STG-барьеры (строки 123-124) **не трогаем**:
STG целиком эталонный, барьер `check_airplanes_dq → check_routes_dq` работает.
- `tests/test_ods_sql_contract.py` строка 6:
На main изменить `SNAPSHOT_ENTITIES` — исключить студенческие ODS:
```python
# Было:
SNAPSHOT_ENTITIES = ("airports", "airplanes", "routes", "seats")
# Стало (main):
SNAPSHOT_ENTITIES = ("airports", "routes")
```
Тесты `test_snapshot_load_scripts_use_truncate` (строка 13) и
`test_snapshot_dq_checks_extra_keys` (строка 22) проверяют паттерны TRUNCATE+INSERT
и `v_extra_keys_count` в SQL-файлах. Заглушки `SELECT 1;` не содержат этих паттернов →
тесты упадут для airplanes/seats. Исключение из `SNAPSHOT_ENTITIES` решает проблему.
Batch resolver тест (строка 30) проверяет DAG-код (не SQL-файлы) — без изменений.
### 11. Прочие документы
- `docs/design/PRD.md` — обновить чеклист (раздел 10)
- `docs/design/bookings_ods_design.md` — комплексное обновление:
- Строка 29, 515, 534: пометить `airplanes` и `seats` как студенческие
(на main — заглушки; STG-уровень — эталон, ODS load/dq — студент)
- Строки 521, 523: скорректировать общий DQ-контракт — на main `airplanes_dq.sql`
и `seats_dq.sql` являются заглушками, а не полноценными RAISE EXCEPTION скриптами
- Строка 620: обновить DAG-граф — routes больше не зависит от airplanes на ODS
- Строка 634: убрать зависимость routes от airplanes в описании FK
- `docs/bookings_to_gp_ods.md` — обновить описание
- `docs/bookings_to_gp_dds.md`:
- Строки 111-113: убрать гарантию отсутствия NULL SK в штатном режиме —
на main student SK (`passenger_sk`, `route_sk`, `airplane_sk`) будут NULL
- Строка 116: обновить DQ-семантику — student SK не блокируют (NOTICE);
airport_sk — порог 1%
- Строка 126: обновить описание lookup (airports через ods.routes)
- `docs/reference/qa-plan.md`:
- Строка 80: «ODS: все 9 таблиц не пустые» — на main `ods.airplanes` и `ods.seats`
будут пустыми by design (студенческие заглушки). Уточнить формулировку.
- `docs/e2e-etl-test-protocol.md` — обновить контекст
---
## Порядок выполнения
### Сейчас (на chore/bookings-etl)
1. **Код**: изменить `fact_flight_sales_load.sql` (lookup через ods.routes) — п.1
2. **Документация**: обновить `analyst_spec.md` — п.6
3. **Документация**: обновить `assignment_design.md` — п.7
4. **Документация**: обновить `db_schema.md` — п.9
5. **Тестирование**: `make test` + ручная проверка SQL
6. **Коммит**
### Этап 3 (валидационный DAG)
- Проектировать с учётом DDL-заглушек
- `validate_stg` секцию **не делать** (STG — эталон)
- `check_dim_routes_scd2` — активный тест (мутация ODS → проверка → откат)
### Этап 4 (подготовка main + solution)
- Fact DQ: ослабить проверки student SK (main only) — п.2
- Routes DQ: убрать airplane RI (main only) — п.3
- DAG-зависимости: убрать airplanes → routes (main only) — п.4
- Заглушки: ODS (airplanes, seats) + DDS (3 dims) + DM (4 vitrines) — п.5
- Тесты: обновить smoke-тесты (main only) — п.10
- Документация: bookings_dds_design, PRD, прочее — п.8, п.11
---
## Верификация
### После п.1 (fact_flight_sales fix)
```bash
make test # smoke-тесты DAG
```
Ручная проверка (при поднятом стенде):
> **Важно:** UPDATE факта не перезаписывает dimension SK (by design, строка 4-5).
> Уже загруженные строки сохранят старые значения `airport_sk` (через dim_routes).
> Для полной проверки новой логики нужен **TRUNCATE + перезагрузка**:
```sql
TRUNCATE dds.fact_flight_sales;
-- Перезапустить DDS DAG (или только task load_dds_fact_flight_sales)
-- Проверка: airport_sk заполнены (через ods.routes), route_sk тоже (dim_routes заполнен на этой ветке)
SELECT
COUNT(*) AS total,
COUNT(departure_airport_sk) AS has_dep_sk,
COUNT(arrival_airport_sk) AS has_arr_sk,
COUNT(route_sk) AS has_route_sk
FROM dds.fact_flight_sales;
```
### После Этапа 4 (main branch)
1. `make up` → `make ddl-gp` → запустить STG+ODS+DDS DAG-и
2. Проверить: `sales_report` содержит данные с корректными аэропортами
3. Проверить: студенческие ODS/DDS-таблицы пусты (заглушки)
4. Fact DQ проходит (student SK = NULL, но DQ не блокирует)
5. Реализовать один студенческий dim → TRUNCATE fact → re-run → проверить SK
---
## Ключевые файлы
| Файл | Роль | Ветка |
|------|------|-------|
| `sql/dds/fact_flight_sales_load.sql` | Главное изменение: airport lookup через ods.routes | обе |
| `sql/dds/fact_flight_sales_dq.sql` | Student SK: EXCEPTION → NOTICE | main only |
| `sql/ods/routes_dq.sql` | Убрать RI airplane_code → ods.airplanes | main only |
| `airflow/dags/bookings_to_gp_ods.py` | DAG: убрать airplanes → routes | main only |
| `tests/test_ods_sql_contract.py` | SNAPSHOT_ENTITIES: убрать airplanes, seats | main only |
| `tests/test_dags_smoke.py` | ODS-барьер airplanes → routes | main only |
| `docs/assignment/analyst_spec.md` | Убрать STG-задания, убрать ods.routes | обе |
| `docs/design/assignment_design.md` | Обновить таблицы эталон/задание | обе |
| `docs/design/bookings_dds_design.md` | Обновить описание fact lookup | обе |
| `docs/design/bookings_ods_design.md` | Убрать зависимость routes от airplanes | main only |
| `docs/design/db_schema.md` | Пометить STG + ods.routes как эталон | обе |
> **STG не трогаем**: `sql/stg/routes_dq.sql`, `bookings_to_gp_stage.py`,
> STG smoke-тесты — без изменений. Весь STG эталонный, зависимости работают штатно.
docs/plans/2026-03-12_validation-dag.md
+999
View File
@@ -0,0 +1,999 @@
# План: Валидационный DAG `bookings_validate`
## Контекст и мотивация
### Проблема
Студент реализует 9 объектов (2 ODS, 3 DDS-измерения, 4 DM-витрины) и **не имеет
автоматической обратной связи** — правильно ли работает его код. Сейчас единственный
способ проверки — ручные SQL-запросы и визуальный контроль данных в таблицах.
Это приводит к типичным ошибкам, которые студент не замечает:
| Ошибка | Где проявляется | Когда обнаруживается |
|--------|-----------------|----------------------|
| NULL в PK | ODS/DDS | Только при построении витрины (неожиданные NULL в агрегатах) |
| Дубли по BK | DDS dim_passengers | Факт раздувается, витрины считают неверно |
| SCD2 не закрывает версии | DDS dim_routes | `valid_to` всегда NULL → point-in-time JOIN перестаёт работать |
| «Дыры» в SCD2-интервалах | DDS dim_routes | `sales_report` теряет строки за «пустые» даты |
| TRUNCATE+INSERT не вытащил все строки | ODS airplanes/seats | DDS-измерение неполное → NULL в факте |
| Витрина пуста при непустом источнике | DM | Видно сразу, но причина непонятна |
Без валидационного DAG студент узнаёт о проблеме **через 2-3 слоя** — на этапе DM-витрин,
где отладка многократно сложнее.
### Решение
**Отдельный DAG `bookings_validate`** — набор проверок, сгруппированных по слоям.
Студент запускает его вручную (trigger в Airflow UI) после реализации заданий.
Каждый таск — одна проверка, с дружелюбным сообщением об ошибке и подсказкой,
что делать дальше.
### Педагогическая ценность
1. **Практика чтения логов Airflow** — студент учится находить ошибки в логах тасков
2. **Инкрементальная обратная связь** — можно запускать после каждого шага, не дожидаясь
реализации всех заданий
3. **Паттерн DQ в пайплайне** — студент видит, как устроены production-проверки качества данных
4. **Активный тест SCD2** — единственный способ убедиться, что SCD2-логика действительно
работает (справочник `routes` в демо-базе статичен)
### Дизайн-решения
| Решение | Обоснование |
|---------|-------------|
| Отдельный DAG (не часть основного пайплайна) | Запускается по желанию, не блокирует основную загрузку |
| `schedule=None` (ручной триггер) | Студент запускает, когда готов проверить свой код |
| `PostgresOperator` + SQL-скрипты | Единый паттерн с основными DAG-ами |
| SQL в `sql/validate/` | Отдельный каталог — валидация не смешивается с DQ пайплайна |
| `DO $$ ... RAISE EXCEPTION ... $$` | Тот же паттерн, что в эталонных DQ-скриптах |
| TaskGroup по слоям | Студент видит, на каком слое проблема |
| Проверки «существует + не пуста» для DM | Минимально необходимо; бизнес-инварианты студент добавит сам в DQ витрин |
---
## Структура DAG
```
bookings_validate
├── validate_ods (TaskGroup)
│ ├── check_ods_airplanes_rowcount
│ ├── check_ods_seats_rowcount
│ ├── check_ods_no_dup_bk
│ └── check_ods_no_null_pks
├── validate_dds (TaskGroup)
│ ├── check_dim_airplanes_exists
│ ├── check_dim_passengers_exists
│ ├── check_dim_passengers_no_dup_bk
│ ├── check_dim_routes_exists
│ ├── check_dim_routes_scd2 ← активный тест!
│ └── check_dim_routes_no_gaps
└── validate_dm (TaskGroup)
├── check_airport_traffic_exists
├── check_route_performance_exists
├── check_monthly_overview_exists
└── check_passenger_loyalty_exists
```
### Зависимости между группами
Нет жёстких зависимостей. Все три группы запускаются параллельно — студент может
реализовать только ODS и увидеть зелёные чеки для `validate_ods`, пока DDS/DM ещё
красные. Это даёт инкрементальную обратную связь.
Внутри каждой группы таски также параллельны (независимые проверки).
---
## Детализация проверок
### ODS: `check_ods_airplanes_rowcount`
**Файл:** `sql/validate/ods_airplanes_rowcount.sql`
**Логика:** ODS загружает один конкретный snapshot-батч из STG (по `_load_id`).
STG — append-only: содержит историю всех загрузок. Поэтому сравнивать ODS
со всей STG некорректно — нужно проверять точное множество BK для того батча,
который ODS фактически загрузил.
Определяем батч: `_load_id` из `ods.airplanes` (единственный, т.к. TRUNCATE+INSERT).
Затем проверяем, что множество BK в ODS = множество BK в STG для этого батча.
```sql
DO $$
DECLARE
v_batch_count BIGINT;
v_batch TEXT;
v_ods_count BIGINT;
v_missing_in_ods BIGINT;
v_extra_in_ods BIGINT;
BEGIN
-- ODS пуста?
SELECT COUNT(*) INTO v_ods_count FROM ods.airplanes;
IF v_ods_count = 0 THEN
RAISE EXCEPTION 'FAILED: ods.airplanes пуста. Реализуйте загрузку: sql/ods/airplanes_load.sql';
END IF;
-- Инвариант: ODS после TRUNCATE+INSERT содержит ровно один _load_id
SELECT COUNT(DISTINCT _load_id) INTO v_batch_count FROM ods.airplanes;
IF v_batch_count <> 1 THEN
RAISE EXCEPTION 'FAILED: ods.airplanes содержит % разных _load_id (ожидается 1 после TRUNCATE+INSERT). Проверьте, что load начинается с TRUNCATE.', v_batch_count;
END IF;
-- Определяем батч, из которого загружена ODS
SELECT DISTINCT _load_id INTO v_batch FROM ods.airplanes;
-- BK есть в STG-батче, но нет в ODS (потеряны при загрузке)
SELECT COUNT(*) INTO v_missing_in_ods
FROM (
SELECT DISTINCT airplane_code FROM stg.airplanes WHERE _load_id = v_batch
) AS stg_bk
WHERE NOT EXISTS (
SELECT 1 FROM ods.airplanes AS o WHERE o.airplane_code = stg_bk.airplane_code
);
IF v_missing_in_ods > 0 THEN
RAISE EXCEPTION 'FAILED: % самолётов из STG-батча (%) отсутствуют в ods.airplanes. Проверьте логику TRUNCATE+INSERT.', v_missing_in_ods, v_batch;
END IF;
-- BK есть в ODS, но нет в STG-батче (откуда взялись?)
SELECT COUNT(*) INTO v_extra_in_ods
FROM (
SELECT DISTINCT airplane_code FROM ods.airplanes
) AS ods_bk
WHERE NOT EXISTS (
SELECT 1 FROM stg.airplanes AS s WHERE s._load_id = v_batch AND s.airplane_code = ods_bk.airplane_code
);
IF v_extra_in_ods > 0 THEN
RAISE EXCEPTION 'FAILED: % самолётов в ods.airplanes отсутствуют в STG-батче (%). Возможно, TRUNCATE не выполнился перед INSERT.', v_extra_in_ods, v_batch;
END IF;
RAISE NOTICE 'PASSED: ods.airplanes содержит % самолётов, множество BK = STG-батч %', v_ods_count, v_batch;
END $$;
```
### ODS: `check_ods_seats_rowcount`
**Файл:** `sql/validate/ods_seats_rowcount.sql`
Аналогичная проверка по составному ключу `(airplane_code, seat_no)` —
точное множество BK из STG-батча (по `_load_id` из `ods.seats`).
### ODS: `check_ods_no_dup_bk`
**Файл:** `sql/validate/ods_no_dup_bk.sql`
**Логика:** Проверить, что в ODS нет дублей по бизнес-ключу. Типичная ошибка
студента — INSERT без предшествующего TRUNCATE, или TRUNCATE забыт при повторном
запуске. Дубли по BK в ODS каскадно ломают DDS (лишние строки в измерениях).
```sql
DO $$
DECLARE
v_dup_airplanes BIGINT;
v_dup_seats BIGINT;
BEGIN
-- airplanes: BK = airplane_code
SELECT COUNT(*) INTO v_dup_airplanes
FROM (
SELECT airplane_code FROM ods.airplanes
GROUP BY airplane_code HAVING COUNT(*) > 1
) AS d;
IF v_dup_airplanes > 0 THEN
RAISE EXCEPTION 'FAILED: ods.airplanes содержит % дублирующихся airplane_code. Проверьте, что load начинается с TRUNCATE.', v_dup_airplanes;
END IF;
-- seats: BK = (airplane_code, seat_no)
SELECT COUNT(*) INTO v_dup_seats
FROM (
SELECT airplane_code, seat_no FROM ods.seats
GROUP BY airplane_code, seat_no HAVING COUNT(*) > 1
) AS d;
IF v_dup_seats > 0 THEN
RAISE EXCEPTION 'FAILED: ods.seats содержит % дублирующихся (airplane_code, seat_no). Проверьте, что load начинается с TRUNCATE.', v_dup_seats;
END IF;
RAISE NOTICE 'PASSED: Нет дублей по BK в ods.airplanes и ods.seats';
END $$;
```
### ODS: `check_ods_no_null_pks`
**Файл:** `sql/validate/ods_no_null_pks.sql`
**Логика:** Проверить, что PK-поля не содержат NULL в обеих студенческих ODS-таблицах.
```sql
DO $$
DECLARE
v_null_airplanes BIGINT;
v_null_seats BIGINT;
BEGIN
-- airplanes: PK = airplane_code
SELECT COUNT(*) INTO v_null_airplanes
FROM ods.airplanes WHERE airplane_code IS NULL;
IF v_null_airplanes > 0 THEN
RAISE EXCEPTION 'FAILED: ods.airplanes содержит % строк с NULL в airplane_code.', v_null_airplanes;
END IF;
-- seats: PK = (airplane_code, seat_no)
SELECT COUNT(*) INTO v_null_seats
FROM ods.seats WHERE airplane_code IS NULL OR seat_no IS NULL;
IF v_null_seats > 0 THEN
RAISE EXCEPTION 'FAILED: ods.seats содержит % строк с NULL в PK (airplane_code, seat_no).', v_null_seats;
END IF;
RAISE NOTICE 'PASSED: PK не содержат NULL в ods.airplanes и ods.seats';
END $$;
```
---
### DDS: `check_dim_airplanes_exists`
**Файл:** `sql/validate/dim_airplanes_exists.sql`
**Логика:**
1. Таблица не пуста
2. Нет дублей по `airplane_bk`
3. Покрытие ODS: все `airplane_code` из `ods.airplanes` есть в `dds.dim_airplanes`
```sql
DO $$
DECLARE
v_count BIGINT;
v_dup BIGINT;
v_missing BIGINT;
BEGIN
SELECT COUNT(*) INTO v_count FROM dds.dim_airplanes;
IF v_count = 0 THEN
RAISE EXCEPTION 'FAILED: dds.dim_airplanes пуста. Реализуйте загрузку: sql/dds/dim_airplanes_load.sql';
END IF;
SELECT COUNT(*) - COUNT(DISTINCT airplane_bk) INTO v_dup FROM dds.dim_airplanes;
IF v_dup > 0 THEN
RAISE EXCEPTION 'FAILED: dds.dim_airplanes содержит % дублей по airplane_bk. Проверьте SCD1-логику (UPSERT).', v_dup;
END IF;
SELECT COUNT(*) INTO v_missing
FROM (SELECT DISTINCT airplane_code FROM ods.airplanes) AS s
WHERE NOT EXISTS (
SELECT 1 FROM dds.dim_airplanes AS d WHERE d.airplane_bk = s.airplane_code
);
IF v_missing > 0 THEN
RAISE EXCEPTION 'FAILED: % самолётов из ods.airplanes отсутствуют в dds.dim_airplanes.', v_missing;
END IF;
RAISE NOTICE 'PASSED: dds.dim_airplanes содержит % строк, покрывает все BK из ODS', v_count;
END $$;
```
### DDS: `check_dim_passengers_exists`
**Файл:** `sql/validate/dim_passengers_exists.sql`
Аналогичная структура: не пуста + нет дублей по `passenger_id` + покрытие
`ods.tickets` (все уникальные `passenger_id` из тикетов есть в измерении).
### DDS: `check_dim_passengers_no_dup_bk`
**Файл:** `sql/validate/dim_passengers_no_dup_bk.sql`
Отдельная проверка дублей `passenger_id` — типичная ошибка студентов при SCD1:
INSERT без проверки EXISTS создаёт дубли. Выделена в отдельный таск для ясности
сообщения об ошибке.
**Примечание:** Эту проверку можно объединить с `check_dim_passengers_exists`.
Отдельный таск оправдан, если хотим дать студенту более точную диагностику.
Решение — на усмотрение реализатора.
### DDS: `check_dim_routes_exists`
**Файл:** `sql/validate/dim_routes_exists.sql`
Таблица не пуста + покрытие ODS (`route_no`) + **не более одной текущей версии
на `route_bk`** (инвариант SCD2: `COUNT(*) WHERE valid_to IS NULL` <= 1 для каждого BK).
Типичная ошибка студента — INSERT без проверки `NOT EXISTS ... AND hashdiff = ...`,
что создаёт дубли текущей версии. Активный SCD2-тест проверяет только один маршрут,
эта проверка ловит проблему глобально.
```sql
DO $$
DECLARE
v_count BIGINT;
v_missing BIGINT;
v_multi_current BIGINT;
BEGIN
SELECT COUNT(*) INTO v_count FROM dds.dim_routes;
IF v_count = 0 THEN
RAISE EXCEPTION 'FAILED: dds.dim_routes пуста. Реализуйте загрузку: sql/dds/dim_routes_load.sql';
END IF;
-- Покрытие ODS
SELECT COUNT(*) INTO v_missing
FROM (SELECT DISTINCT route_no FROM ods.routes) AS s
WHERE NOT EXISTS (
SELECT 1 FROM dds.dim_routes AS d WHERE d.route_bk = s.route_no
);
IF v_missing > 0 THEN
RAISE EXCEPTION 'FAILED: % маршрутов из ods.routes отсутствуют в dds.dim_routes.', v_missing;
END IF;
-- Инвариант SCD2: не более одной текущей версии на route_bk
SELECT COUNT(*) INTO v_multi_current
FROM (
SELECT route_bk FROM dds.dim_routes
WHERE valid_to IS NULL
GROUP BY route_bk HAVING COUNT(*) > 1
) AS d;
IF v_multi_current > 0 THEN
RAISE EXCEPTION E'FAILED: % маршрутов имеют более одной текущей версии (valid_to IS NULL).\n'
'Подсказка: при INSERT новой версии проверяйте NOT EXISTS ... AND hashdiff = ...\n'
'чтобы не создавать дубликат, если hashdiff не изменился.', v_multi_current;
END IF;
RAISE NOTICE 'PASSED: dds.dim_routes содержит % строк, покрывает ODS, по одной текущей версии на маршрут', v_count;
END $$;
```
### DDS: `check_dim_routes_scd2` (АКТИВНЫЙ ТЕСТ)
**Файл:** `sql/validate/dim_routes_scd2.sql`
Это ключевая проверка плана. Справочник `bookings.routes` в демо-базе **статичен** —
маршруты не меняются между запусками генератора. При обычном прогоне пайплайна
студент **никогда не увидит**, как SCD2 закрывает старую версию и создаёт новую.
**Алгоритм активного теста:**
```
1. Бэкап: CREATE TABLE _validate_bk_ods_routes AS SELECT * FROM ods.routes;
CREATE TABLE _validate_bk_dim_routes AS SELECT * FROM dds.dim_routes;
(обычные таблицы — TEMP не сохраняются между тасками Airflow)
2. Мутация: Выбрать один маршрут из ods.routes (WHERE departure_time IS NOT NULL LIMIT 1).
Сохранить его route_no в служебную таблицу _validate_scd2_target.
Обновить в ods.routes его departure_time на +1 час.
3. Запуск студенческого кода: PostgresOperator(sql="dds/dim_routes_load.sql")
4. Проверки (все 5):
a) Ровно 2 версии тестового маршрута (было 1, стало 2)
b) Старая версия закрыта: valid_to IS NOT NULL
c) Ровно 1 открытая версия: valid_to IS NULL
d) hashdiff старой ≠ hashdiff новой (мутация отразилась в хеше)
e) Нет «дыры»: старая.valid_to = новая.valid_from
5. Откат (с проверкой существования бэкапов через to_regclass):
IF _validate_bk_ods_routes exists: TRUNCATE ods.routes + INSERT FROM backup;
IF _validate_bk_dim_routes exists: TRUNCATE dds.dim_routes + INSERT FROM backup;
DROP TABLE IF EXISTS _validate_bk_*, _validate_scd2_target;
```
**Реализация:** Этот таск **не может быть простым PostgresOperator** с одним SQL,
потому что шаг 3 — это вызов студенческого SQL-скрипта *внутри* теста. Варианты:
**Выбранный подход: цепочка тасков.**
- Jinja-шаблоны работают из коробки
- Каждый шаг прозрачен в Airflow UI
- `trigger_rule="all_done"` на restore гарантирует откат
- Студент видит в UI, на каком именно шаге проблема
### DDS: `check_dim_routes_no_gaps`
**Файл:** `sql/validate/dim_routes_no_gaps.sql`
**Логика:** Для каждого `route_bk` с несколькими версиями проверить, что
`valid_to` предыдущей версии = `valid_from` следующей (полуоткрытый интервал
`[valid_from, valid_to)` без «дыр»).
**Важно:** SCD2-загрузка поддерживает «исчезнувшие» маршруты (`dim_routes_load.sql`,
Statement 1.1): если маршрут пропал из ODS, его текущая версия закрывается
(`valid_to = CURRENT_DATE`), но новая **не вставляется**. Это корректное поведение —
у такого маршрута `valid_to IS NOT NULL` и `next_valid_from IS NULL`. Проверка
должна считать «дырой» только случаи, когда следующая версия **существует**,
но `valid_to ≠ next_valid_from`.
```sql
DO $$
DECLARE
v_gaps BIGINT;
BEGIN
SELECT COUNT(*) INTO v_gaps
FROM (
SELECT route_bk, valid_to,
LEAD(valid_from) OVER (PARTITION BY route_bk ORDER BY valid_from) AS next_valid_from
FROM dds.dim_routes
) AS t
WHERE valid_to IS NOT NULL
AND next_valid_from IS NOT NULL -- следующая версия существует (не «исчезнувший» маршрут)
AND valid_to <> next_valid_from;
IF v_gaps > 0 THEN
RAISE EXCEPTION 'FAILED: В dds.dim_routes найдено % «дыр» между версиями SCD2. valid_to старой версии должен совпадать с valid_from новой.', v_gaps;
END IF;
RAISE NOTICE 'PASSED: Нет «дыр» в SCD2-версиях dim_routes';
END $$;
```
---
### DM: `check_{vitrine}_exists` (4 таска)
**Файлы:**
- `sql/validate/airport_traffic_exists.sql`
- `sql/validate/route_performance_exists.sql`
- `sql/validate/monthly_overview_exists.sql`
- `sql/validate/passenger_loyalty_exists.sql`
**Логика:** Минимальная проверка — таблица не пуста + нет NULL в ключевых полях.
Пример для `airport_traffic`:
```sql
DO $$
DECLARE
v_count BIGINT;
v_null_pk BIGINT;
BEGIN
SELECT COUNT(*) INTO v_count FROM dm.airport_traffic;
IF v_count = 0 THEN
RAISE EXCEPTION 'FAILED: dm.airport_traffic пуста. Реализуйте загрузку: sql/dm/airport_traffic_load.sql';
END IF;
SELECT COUNT(*) INTO v_null_pk
FROM dm.airport_traffic
WHERE traffic_date IS NULL OR airport_sk IS NULL;
IF v_null_pk > 0 THEN
RAISE EXCEPTION 'FAILED: dm.airport_traffic содержит % строк с NULL в ключе (traffic_date, airport_sk).', v_null_pk;
END IF;
RAISE NOTICE 'PASSED: dm.airport_traffic содержит % строк', v_count;
END $$;
```
Для `route_performance` ключ — `route_bk`, для `monthly_overview` — `(year_actual, month_actual, airplane_sk)`,
для `passenger_loyalty` — `passenger_sk`.
---
## Файлы для создания
### SQL-скрипты (`sql/validate/`)
| # | Файл | Описание |
|---|------|----------|
| 1 | `ods_airplanes_rowcount.sql` | BK coverage: ODS vs STG-батч |
| 2 | `ods_seats_rowcount.sql` | BK coverage: ODS vs STG-батч |
| 3 | `ods_no_dup_bk.sql` | Нет дублей по BK в ODS (airplanes + seats) |
| 4 | `ods_no_null_pks.sql` | NULL в PK обеих ODS-таблиц |
| 5 | `dim_airplanes_exists.sql` | Не пуста + нет дублей BK + покрытие ODS |
| 6 | `dim_passengers_exists.sql` | Не пуста + нет дублей BK + покрытие ODS |
| 7 | `dim_passengers_no_dup_bk.sql` | Дубли `passenger_id` (отдельная диагностика) |
| 8 | `dim_routes_exists.sql` | Не пуста + покрытие ODS + не более 1 текущей версии на BK |
| 9 | `dim_routes_scd2_backup.sql` | Бэкап ods.routes + dds.dim_routes |
| 10 | `dim_routes_scd2_mutate.sql` | Мутация тестового маршрута + сохранение route_no |
| 11 | `dim_routes_scd2_check.sql` | 5 проверок SCD2 после загрузки |
| 12 | `dim_routes_scd2_restore.sql` | Безопасный откат данных из бэкапа |
| 13 | `dim_routes_no_gaps.sql` | Проверка SCD2-интервалов (без «исчезнувших») |
| 14 | `airport_traffic_exists.sql` | Не пуста + NULL в ключе |
| 15 | `route_performance_exists.sql` | Не пуста + NULL в ключе |
| 16 | `monthly_overview_exists.sql` | Не пуста + NULL в ключе |
| 17 | `passenger_loyalty_exists.sql` | Не пуста + NULL в ключе |
### DAG-файл
| # | Файл | Описание |
|---|------|----------|
| 18 | `airflow/dags/bookings_validate.py` | DAG с TaskGroup по слоям |
### Тест
| # | Файл | Описание |
|---|------|----------|
| 19 | Дополнение `tests/test_dags_smoke.py` | Smoke-тест структуры DAG |
---
## DAG-файл: `bookings_validate.py`
```python
"""
Валидационный DAG: самопроверка студенческих заданий.
Запускается вручную в Airflow UI после реализации заданий.
Таски сгруппированы по слоям — студент видит, где именно проблема.
"""
from datetime import timedelta
from logging import getLogger
import pendulum
from airflow import DAG
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.utils.task_group import TaskGroup
GREENPLUM_CONN_ID = "greenplum_conn"
log = getLogger(__name__)
default_args = {
"owner": "airflow",
"retries": 0, # Без ретраев — студент должен увидеть ошибку сразу
"retry_delay": timedelta(seconds=10),
}
with DAG(
dag_id="bookings_validate",
start_date=pendulum.datetime(2017, 1, 1, tz="UTC"),
schedule=None, # Только ручной запуск
catchup=False,
max_active_runs=1,
template_searchpath="/sql",
default_args=default_args,
tags=["demo", "bookings", "greenplum", "validate"],
description="Валидация студенческих заданий: ODS, DDS, DM",
) as dag:
with TaskGroup("validate_ods") as validate_ods:
check_ods_airplanes = PostgresOperator(
task_id="check_ods_airplanes_rowcount",
postgres_conn_id=GREENPLUM_CONN_ID,
sql="validate/ods_airplanes_rowcount.sql",
)
check_ods_seats = PostgresOperator(
task_id="check_ods_seats_rowcount",
postgres_conn_id=GREENPLUM_CONN_ID,
sql="validate/ods_seats_rowcount.sql",
)
check_ods_dup_bk = PostgresOperator(
task_id="check_ods_no_dup_bk",
postgres_conn_id=GREENPLUM_CONN_ID,
sql="validate/ods_no_dup_bk.sql",
)
check_ods_pks = PostgresOperator(
task_id="check_ods_no_null_pks",
postgres_conn_id=GREENPLUM_CONN_ID,
sql="validate/ods_no_null_pks.sql",
)
with TaskGroup("validate_dds") as validate_dds:
check_dim_airplanes = PostgresOperator(
task_id="check_dim_airplanes_exists",
postgres_conn_id=GREENPLUM_CONN_ID,
sql="validate/dim_airplanes_exists.sql",
)
check_dim_passengers = PostgresOperator(
task_id="check_dim_passengers_exists",
postgres_conn_id=GREENPLUM_CONN_ID,
sql="validate/dim_passengers_exists.sql",
)
check_dim_passengers_dup = PostgresOperator(
task_id="check_dim_passengers_no_dup_bk",
postgres_conn_id=GREENPLUM_CONN_ID,
sql="validate/dim_passengers_no_dup_bk.sql",
)
check_dim_routes = PostgresOperator(
task_id="check_dim_routes_exists",
postgres_conn_id=GREENPLUM_CONN_ID,
sql="validate/dim_routes_exists.sql",
)
# SCD2 активный тест: цепочка backup → mutate → load → check → restore
scd2_backup = PostgresOperator(
task_id="scd2_backup",
postgres_conn_id=GREENPLUM_CONN_ID,
sql="validate/dim_routes_scd2_backup.sql",
)
scd2_mutate = PostgresOperator(
task_id="scd2_mutate",
postgres_conn_id=GREENPLUM_CONN_ID,
sql="validate/dim_routes_scd2_mutate.sql",
)
scd2_run_load = PostgresOperator(
task_id="scd2_run_student_load",
postgres_conn_id=GREENPLUM_CONN_ID,
sql="dds/dim_routes_load.sql", # студенческий load-скрипт!
)
scd2_check = PostgresOperator(
task_id="scd2_check",
postgres_conn_id=GREENPLUM_CONN_ID,
sql="validate/dim_routes_scd2_check.sql",
)
scd2_restore = PostgresOperator(
task_id="scd2_restore",
postgres_conn_id=GREENPLUM_CONN_ID,
sql="validate/dim_routes_scd2_restore.sql",
trigger_rule="all_done", # Откат ВСЕГДА, даже если check упал
)
check_dim_routes_gaps = PostgresOperator(
task_id="check_dim_routes_no_gaps",
postgres_conn_id=GREENPLUM_CONN_ID,
sql="validate/dim_routes_no_gaps.sql",
)
# SCD2 цепочка
scd2_backup >> scd2_mutate >> scd2_run_load >> scd2_check >> scd2_restore
# no_gaps запускается после restore (на чистых данных)
scd2_restore >> check_dim_routes_gaps
with TaskGroup("validate_dm") as validate_dm:
check_airport_traffic = PostgresOperator(
task_id="check_airport_traffic_exists",
postgres_conn_id=GREENPLUM_CONN_ID,
sql="validate/airport_traffic_exists.sql",
)
check_route_performance = PostgresOperator(
task_id="check_route_performance_exists",
postgres_conn_id=GREENPLUM_CONN_ID,
sql="validate/route_performance_exists.sql",
)
check_monthly_overview = PostgresOperator(
task_id="check_monthly_overview_exists",
postgres_conn_id=GREENPLUM_CONN_ID,
sql="validate/monthly_overview_exists.sql",
)
check_passenger_loyalty = PostgresOperator(
task_id="check_passenger_loyalty_exists",
postgres_conn_id=GREENPLUM_CONN_ID,
sql="validate/passenger_loyalty_exists.sql",
)
```
---
## Активный тест SCD2: детализация SQL
### `dim_routes_scd2_backup.sql`
```sql
-- Бэкап текущего состояния перед тестом SCD2.
-- Используем обычные таблицы (не TEMP) — между тасками Airflow
-- TEMP-таблицы не сохраняются (каждый таск = отдельная транзакция).
DROP TABLE IF EXISTS _validate_bk_ods_routes;
CREATE TABLE _validate_bk_ods_routes AS SELECT * FROM ods.routes;
DROP TABLE IF EXISTS _validate_bk_dim_routes;
CREATE TABLE _validate_bk_dim_routes AS SELECT * FROM dds.dim_routes;
-- Cleanup служебной таблицы от предыдущего запуска (на случай если restore не доехал)
DROP TABLE IF EXISTS _validate_scd2_target;
```
### `dim_routes_scd2_mutate.sql`
```sql
-- Мутация: сдвигаем departure_time у одного маршрута на 1 час.
-- Это должно изменить hashdiff → SCD2 должен закрыть старую версию.
--
-- Сохраняем route_no тестового маршрута в служебную таблицу _validate_scd2_target,
-- чтобы check-скрипт точно знал, какой маршрут проверять (а не угадывал по побочным эффектам).
DO $$
DECLARE
v_route TEXT;
v_old_time TIME;
BEGIN
-- Берём первый маршрут, у которого departure_time заполнен
SELECT route_no, departure_time
INTO v_route, v_old_time
FROM ods.routes
WHERE departure_time IS NOT NULL
ORDER BY route_no
LIMIT 1;
IF v_route IS NULL THEN
RAISE EXCEPTION 'FAILED: ods.routes пуста или нет маршрутов с departure_time. Загрузите STG→ODS перед проверкой.';
END IF;
-- Запоминаем тестовый маршрут в служебную таблицу
DROP TABLE IF EXISTS _validate_scd2_target;
CREATE TABLE _validate_scd2_target AS
SELECT v_route AS route_no;
-- Сдвигаем время на 1 час у всех записей этого маршрута
UPDATE ods.routes
SET departure_time = departure_time + INTERVAL '1 hour'
WHERE route_no = v_route;
RAISE NOTICE 'SCD2 TEST: маршрут % — departure_time сдвинут с % на %',
v_route, v_old_time, v_old_time + INTERVAL '1 hour';
END $$;
```
### `dim_routes_scd2_check.sql`
```sql
-- Проверяем, что SCD2-логика студента сработала корректно.
-- Читаем route_no тестового маршрута из служебной таблицы _validate_scd2_target
-- (создана на шаге mutate), а не угадываем по побочным эффектам.
DO $$
DECLARE
v_route TEXT;
v_version_count BIGINT;
v_closed_count BIGINT;
v_open_count BIGINT;
v_old_hash TEXT;
v_new_hash TEXT;
v_gap_count BIGINT;
BEGIN
-- Читаем тестовый маршрут из служебной таблицы
SELECT route_no INTO v_route FROM _validate_scd2_target LIMIT 1;
IF v_route IS NULL THEN
RAISE EXCEPTION 'FAILED: служебная таблица _validate_scd2_target пуста. Шаг mutate не выполнился?';
END IF;
-- Если dim_routes пуста — load не запустился
IF NOT EXISTS (SELECT 1 FROM dds.dim_routes WHERE route_bk = v_route) THEN
RAISE EXCEPTION E'FAILED: dds.dim_routes не содержит маршрут % после запуска load.\n'
'Проверьте sql/dds/dim_routes_load.sql.', v_route;
END IF;
-- Проверка a: Ровно 2 версии тестового маршрута (было 1, стало 2 после мутации)
SELECT COUNT(*) INTO v_version_count
FROM dds.dim_routes WHERE route_bk = v_route;
IF v_version_count < 2 THEN
RAISE EXCEPTION E'FAILED: Маршрут % — найдена % версия (ожидается 2: старая закрытая + новая открытая).\n'
'SCD2 должен был создать новую версию после изменения departure_time.', v_route, v_version_count;
END IF;
IF v_version_count > 2 THEN
RAISE EXCEPTION E'FAILED: Маршрут % — найдено % версий (ожидается 2).\n'
'Возможно, load создаёт лишние дубликаты. Проверьте условие NOT EXISTS при INSERT.', v_route, v_version_count;
END IF;
-- Проверка b: Старая версия закрыта (valid_to IS NOT NULL)
SELECT COUNT(*) INTO v_closed_count
FROM dds.dim_routes WHERE route_bk = v_route AND valid_to IS NOT NULL;
IF v_closed_count = 0 THEN
RAISE EXCEPTION E'FAILED: Маршрут % — SCD2 не закрыл старую версию (valid_to IS NULL у всех версий).\n'
'Подсказка: hashdiff изменился (departure_time сдвинут на 1 час),\n'
'но ваш load-скрипт не обнаружил это изменение.\n'
'Проверьте:\n'
' 1. Формулу hashdiff — включает ли она departure_time?\n'
' 2. Логику сравнения hashdiff (UPDATE ... SET valid_to = CURRENT_DATE WHERE hashdiff <> новый_hashdiff)', v_route;
END IF;
-- Проверка c: Новая версия открыта (valid_to IS NULL)
SELECT COUNT(*) INTO v_open_count
FROM dds.dim_routes WHERE route_bk = v_route AND valid_to IS NULL;
IF v_open_count <> 1 THEN
RAISE EXCEPTION E'FAILED: Маршрут % — ожидается ровно 1 открытая версия (valid_to IS NULL), найдено %.\n'
'Подсказка: SCD2 должен вставить новую строку с valid_to = NULL.', v_route, v_open_count;
END IF;
-- Проверка d: hashdiff старой ≠ hashdiff новой (мутация действительно отразилась)
SELECT hashdiff INTO v_old_hash
FROM dds.dim_routes WHERE route_bk = v_route AND valid_to IS NOT NULL
ORDER BY valid_from DESC LIMIT 1;
SELECT hashdiff INTO v_new_hash
FROM dds.dim_routes WHERE route_bk = v_route AND valid_to IS NULL;
IF v_old_hash = v_new_hash THEN
RAISE EXCEPTION E'FAILED: Маршрут % — hashdiff старой и новой версий совпадают.\n'
'Мутация сдвинула departure_time на 1 час, но hashdiff не изменился.\n'
'Проверьте, что departure_time входит в формулу hashdiff.', v_route;
END IF;
-- Проверка e: Нет «дыры» между valid_to старой и valid_from новой
SELECT COUNT(*) INTO v_gap_count
FROM dds.dim_routes AS old_v
JOIN dds.dim_routes AS new_v
ON old_v.route_bk = new_v.route_bk
WHERE old_v.route_bk = v_route
AND old_v.valid_to IS NOT NULL
AND new_v.valid_to IS NULL
AND old_v.valid_to <> new_v.valid_from;
IF v_gap_count > 0 THEN
RAISE EXCEPTION E'FAILED: Маршрут % — «дыра» между версиями:\n'
'valid_to старой ≠ valid_from новой.\n'
'Подсказка: полуоткрытый интервал [valid_from, valid_to).\n'
'valid_from новой версии должен = valid_to старой (обычно CURRENT_DATE).', v_route;
END IF;
RAISE NOTICE 'PASSED: SCD2 корректен для маршрута %. 2 версии, hashdiff различаются, «дыр» нет.', v_route;
END $$;
```
### `dim_routes_scd2_restore.sql`
```sql
-- Откат данных после теста SCD2.
-- Выполняется ВСЕГДА (trigger_rule="all_done"), даже если check упал.
--
-- Безопасность: если backup-шаг не создал таблицы (сбой на backup),
-- откат НЕ трогает live-данные — просто чистит служебные таблицы.
-- Это гарантирует, что restore никогда не сломает ods.routes / dds.dim_routes.
DO $$
DECLARE
v_has_ods_backup BOOLEAN;
v_has_dim_backup BOOLEAN;
BEGIN
-- Проверяем существование backup-таблиц через to_regclass
-- (ищет по search_path — совпадает с тем, как CREATE TABLE их создал)
v_has_ods_backup := to_regclass('_validate_bk_ods_routes') IS NOT NULL;
v_has_dim_backup := to_regclass('_validate_bk_dim_routes') IS NOT NULL;
-- Восстанавливаем ods.routes только если бэкап существует
IF v_has_ods_backup THEN
TRUNCATE ods.routes;
INSERT INTO ods.routes SELECT * FROM _validate_bk_ods_routes;
RAISE NOTICE 'RESTORE: ods.routes восстановлена из бэкапа';
ELSE
RAISE NOTICE 'RESTORE: бэкап ods.routes не найден — пропускаем (backup-шаг не завершился?)';
END IF;
-- Восстанавливаем dds.dim_routes только если бэкап существует
IF v_has_dim_backup THEN
TRUNCATE dds.dim_routes;
INSERT INTO dds.dim_routes SELECT * FROM _validate_bk_dim_routes;
RAISE NOTICE 'RESTORE: dds.dim_routes восстановлена из бэкапа';
ELSE
RAISE NOTICE 'RESTORE: бэкап dds.dim_routes не найден — пропускаем';
END IF;
END $$;
-- Cleanup служебных таблиц (безусловно, IF EXISTS)
DROP TABLE IF EXISTS _validate_bk_ods_routes;
DROP TABLE IF EXISTS _validate_bk_dim_routes;
DROP TABLE IF EXISTS _validate_scd2_target;
-- Учебный комментарий: Мы восстанавливаем данные из бэкапа, чтобы тест
-- не оставлял «мусорных» версий в dim_routes. Это стандартный паттерн
-- для интеграционных тестов: setup → act → assert → teardown.
-- IF EXISTS проверки гарантируют, что restore безопасен при любом сценарии сбоя.
```
---
## Smoke-тест DAG (дополнение `test_dags_smoke.py`)
Добавить новый тест-класс:
```python
class TestBookingsValidate:
"""Smoke-тесты DAG bookings_validate."""
def test_dag_loads(self):
dag = _load_dag("airflow.dags.bookings_validate")
assert dag is not None
def test_expected_tasks(self):
dag = _load_dag("airflow.dags.bookings_validate")
expected = {
# ODS
"validate_ods.check_ods_airplanes_rowcount",
"validate_ods.check_ods_seats_rowcount",
"validate_ods.check_ods_no_dup_bk",
"validate_ods.check_ods_no_null_pks",
# DDS
"validate_dds.check_dim_airplanes_exists",
"validate_dds.check_dim_passengers_exists",
"validate_dds.check_dim_passengers_no_dup_bk",
"validate_dds.check_dim_routes_exists",
"validate_dds.scd2_backup",
"validate_dds.scd2_mutate",
"validate_dds.scd2_run_student_load",
"validate_dds.scd2_check",
"validate_dds.scd2_restore",
"validate_dds.check_dim_routes_no_gaps",
# DM
"validate_dm.check_airport_traffic_exists",
"validate_dm.check_route_performance_exists",
"validate_dm.check_monthly_overview_exists",
"validate_dm.check_passenger_loyalty_exists",
}
assert expected.issubset(dag.task_dict.keys())
def test_scd2_chain(self):
"""SCD2 цепочка backup → mutate → load → check → restore."""
dag = _load_dag("airflow.dags.bookings_validate")
_assert_direct_edge(dag, "validate_dds.scd2_backup", "validate_dds.scd2_mutate")
_assert_direct_edge(dag, "validate_dds.scd2_mutate", "validate_dds.scd2_run_student_load")
_assert_direct_edge(dag, "validate_dds.scd2_run_student_load", "validate_dds.scd2_check")
_assert_direct_edge(dag, "validate_dds.scd2_check", "validate_dds.scd2_restore")
def test_no_gaps_after_restore(self):
"""no_gaps должен выполняться после restore (на чистых данных)."""
dag = _load_dag("airflow.dags.bookings_validate")
_assert_direct_edge(dag, "validate_dds.scd2_restore", "validate_dds.check_dim_routes_no_gaps")
def test_restore_trigger_rule(self):
"""restore должен выполняться всегда (all_done), даже если check упал."""
dag = _load_dag("airflow.dags.bookings_validate")
restore_task = dag.task_dict["validate_dds.scd2_restore"]
assert restore_task.trigger_rule == "all_done"
```
---
## Порядок реализации
1. Создать каталог `sql/validate/`
2. Написать SQL-скрипты (17 файлов) — начать с простых (ODS, DM), затем DDS, затем SCD2
3. Написать DAG `bookings_validate.py`
4. Дополнить `tests/test_dags_smoke.py`
5. `make test` — smoke-тесты проходят
6. Ручная проверка на стенде (если поднят):
- Trigger DAG в Airflow UI
- Все ODS/DDS/DM зелёные (на solution-ветке)
- SCD2 активный тест: backup → mutate → load → check (зелёный) → restore
7. Коммит
---
## Верификация
### Автоматическая
```bash
make test # smoke-тест DAG-структуры
```
### Ручная (на стенде)
1. `make up && make ddl-gp` → запустить STG → ODS → DDS → DM
2. Trigger `bookings_validate` в Airflow UI
3. Проверить: все таски зелёные
4. Проверить SCD2: в логах `scd2_check` видно `PASSED: SCD2 корректен для маршрута ...`
5. Проверить restore: `ods.routes` и `dds.dim_routes` не изменились после теста
### Сценарий «студент ещё не реализовал»
На main-ветке (с DDL-заглушками):
- ODS-таски: FAILED (таблицы пусты) → дружелюбное сообщение
- DDS-таски: FAILED → сообщение «Реализуйте загрузку»
- DM-таски: FAILED → сообщение «Реализуйте загрузку»
- SCD2-тест: `scd2_run_student_load` пройдёт (заглушка `SELECT 1;` — валидный SQL),
но `scd2_check` упадёт (dim_routes не обновилась, версий < 2)
→ `scd2_restore` всё равно выполнится (`trigger_rule="all_done"`)
---
## Открытые вопросы
1. **`dim_passengers_no_dup_bk` — отдельный таск или объединить с `exists`?**
Отдельный таск даёт точнее диагностику, но увеличивает число тасков.
Рекомендация: объединить проверки в один файл `dim_passengers_exists.sql`
(как сделано для `dim_airplanes_exists`).
2. **Нужна ли проверка `total_seats` в `dim_airplanes`?**
`total_seats` вычисляется агрегацией из `ods.seats`. Если студент забудет
этот JOIN — поле будет NULL. Можно добавить `SELECT COUNT(*) WHERE total_seats IS NULL`.
Рекомендация: добавить в `dim_airplanes_exists.sql`.
3. **Бэкап SCD2: обычные таблицы vs TEMP?**
Каждый таск Airflow — отдельная транзакция → TEMP-таблицы не сохраняются.
Используем обычные таблицы с префиксом `_validate_bk_`. Риск: если DAG упадёт
между backup и restore, таблицы останутся. Cleanup: `scd2_restore` делает
`DROP TABLE IF EXISTS`, поэтому при следующем запуске проблем не будет.
---
## Ключевые файлы
| Файл | Роль |
|------|------|
| `airflow/dags/bookings_validate.py` | DAG: TaskGroup по слоям |
| `sql/validate/*.sql` | 17 SQL-скриптов проверок |
| `sql/dds/dim_routes_load.sql` | Студенческий load (вызывается из SCD2-теста) |
| `tests/test_dags_smoke.py` | Smoke-тесты DAG-структуры |
| `docs/design/assignment_design.md` | Дизайн (секция 4 — источник требований) |