Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f1745e01be | ||
|
|
d105cb75df | ||
|
|
7bf61d8f74 | ||
|
|
999a8d08bf | ||
|
|
d3f2f42e38 |
No files matched your search
@@ -74,6 +74,11 @@
|
||||
- `make clean` удаляет тома — предупреждайте студентов, что данные пропадут.
|
||||
|
||||
## Для агента (особенности аудитории)
|
||||
|
||||
При создании и пересмотре лабораторных сначала прочитайте
|
||||
[развитие лабораторных](docs/design/labs_development.md): аудитория,
|
||||
принятые решения и критерии заданий.
|
||||
|
||||
- Пишите простыми словами. Добавляйте короткие комментарии к нетривиальной логике.
|
||||
- Избегайте больших рефакторингов и сложных паттернов — студенты только начинают.
|
||||
- Ошибки и логи — дружелюбные и понятные (лучше с подсказкой «что сделать дальше»).
|
||||
|
||||
@@ -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)"
|
||||
|
||||
@@ -1,5 +1,9 @@
|
||||
# airflow-dwh-gp-lab
|
||||
|
||||
> Этот стенд можно использовать самостоятельно, но он также является курсовой работой
|
||||
> плана обучения [Data Engineering Roadmap](https://github.com/dementev-dev/de-roadmap).
|
||||
> Нужна помощь? Обратитесь к автору материалов — ментору [@dementev_dev](https://t.me/dementev_dev).
|
||||
|
||||
Учебный стенд для лабораторных по Data Engineering: **Airflow** оркестрирует загрузку данных из демо‑БД
|
||||
**bookings** (Postgres) в **Greenplum**.
|
||||
|
||||
@@ -95,6 +99,12 @@ SELECT COUNT(*) FROM dm.route_performance;
|
||||
|
||||
Подробнее про логику DAG и проверки — `docs/bookings_to_gp_stage.md`.
|
||||
|
||||
7) Изучите готовую реализацию:
|
||||
|
||||
Весь пайплайн `STG → ODS → DDS → DM` на этой ветке полностью реализован.
|
||||
Переходите к [навигатору по реализации](docs/assignment/README.md) —
|
||||
там описана структура, ключевые паттерны и рекомендуемый порядок изучения.
|
||||
|
||||
## DAG-и в стенде
|
||||
|
||||
Основные (для потока bookings → DWH):
|
||||
@@ -102,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.
|
||||
|
||||
@@ -70,13 +70,13 @@
|
||||
> Финальный этап: всё готово и протестировано, теперь раскладываем по веткам.
|
||||
> План: `docs/plans/2026-03-12_main-solution-split.md`
|
||||
|
||||
- [ ] Смержить chore/bookings-etl → main
|
||||
- [ ] Общие правки на main (документация, TODO.md)
|
||||
- [ ] Создать ветку solution (снимок полного эталона)
|
||||
- [ ] Main-only правки: заглушки, ослабление DQ, адаптация тестов
|
||||
- [ ] Очистка docs на main (удалить внутренние документы)
|
||||
- [ ] Верификация main (`make test`, `make lint`)
|
||||
- [ ] Верификация solution (`make test`)
|
||||
- [x] Смержить chore/bookings-etl → main
|
||||
- [x] Общие правки на main (документация, TODO.md)
|
||||
- [x] Создать ветку solution (снимок полного эталона)
|
||||
- [x] Main-only правки: заглушки, ослабление DQ, адаптация тестов
|
||||
- [x] Очистка docs на main (удалить внутренние документы)
|
||||
- [x] Верификация main (`make test`, `make lint`)
|
||||
- [x] Верификация solution (`make test`)
|
||||
|
||||
---
|
||||
|
||||
|
||||
@@ -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
|
||||
@@ -23,6 +23,7 @@
|
||||
- [Архитектурные решения (ADR)](design/architecture_review.md)
|
||||
- [PRD: стратегия курсовой](design/PRD.md)
|
||||
- [Дизайн задания](design/assignment_design.md)
|
||||
- [Развитие лабораторных: аудитория, решения и критерии заданий](design/labs_development.md)
|
||||
|
||||
## Справочники (`reference/`)
|
||||
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
# План: Онбординг студента + маркетинг (v4)
|
||||
|
||||
> Статус: выполнен (2026-03-13)
|
||||
|
||||
## Контекст
|
||||
|
||||
Ментор опционален. Студент, клонировав main, проходит «Быстрый старт» — и застревает:
|
||||
нет явного «что дальше», ссылка на задания спрятана, validate DAG не объяснён, нет маркетинга.
|
||||
На main есть противоречия: docs и DAG-docstrings говорят «все реализовано»,
|
||||
фактически стоят заглушки `SELECT 1;`.
|
||||
|
||||
## Что было сделано
|
||||
|
||||
### На main (10 файлов)
|
||||
|
||||
1. **README.md** — маркетинг-баннер, шаг 6 (только эталонные таблицы), шаг 7 → задания.
|
||||
2. **docs/assignment/README.md** — полный гид студента (эталон → ТЗ → заглушки → validate).
|
||||
3. **docs/assignment/analyst_spec.md** — DAG-интеграция: «таски уже подключены, DAG менять не нужно».
|
||||
4. **docs/bookings_to_gp_dm.md** — пометки заглушек, адаптация секций проверки.
|
||||
5. **docs/bookings_to_gp_ods.md** — пометки заглушек airplanes/seats.
|
||||
6. **docs/bookings_to_gp_dds.md** — пометки заглушек dim_routes/dim_passengers/dim_airplanes.
|
||||
7-10. **4 DAG docstrings** — эталон vs задания (заглушки).
|
||||
|
||||
### На solution (follow-up)
|
||||
|
||||
- README.md — маркетинг-баннер + шаг 7 (solution-specific: «Изучите готовую реализацию»).
|
||||
- docs/assignment/README.md — навигационный гид по всем паттернам (ODS → DDS → DM).
|
||||
- Этот план архивирован в docs/archive/.
|
||||
|
||||
## Верификация
|
||||
|
||||
- `make test` — passed на обеих ветках.
|
||||
- `make lint` — clean на обеих ветках.
|
||||
- Grep «заглушки/SELECT 1» в README.md и docs/assignment/README.md на solution — не найдено.
|
||||
@@ -1,17 +1,50 @@
|
||||
# Учебные задания
|
||||
|
||||
> Это ветка `solution` — здесь всё уже реализовано.
|
||||
> Используйте её как справочник, а задания выполняйте на ветке `main`.
|
||||
|
||||
## Навигатор по реализации
|
||||
|
||||
На этой ветке полностью работает пайплайн `STG → ODS → DDS → DM`.
|
||||
Ниже — рекомендуемый порядок изучения: от простых паттернов к сложным.
|
||||
|
||||
### ODS: загрузка из STG
|
||||
|
||||
| Паттерн | Файл | Что посмотреть |
|
||||
|---------|------|----------------|
|
||||
| TRUNCATE + INSERT (snapshot-справочник) | `sql/ods/airports_load.sql` | Фильтрация по `stg_batch_id`, дедупликация через `ROW_NUMBER()` |
|
||||
| SCD1 UPSERT (транзакционные данные) | `sql/ods/bookings_load.sql` | TEMP TABLE → UPDATE (`IS DISTINCT FROM`) → INSERT, HWM |
|
||||
|
||||
### DDS: измерения и факт
|
||||
|
||||
| Паттерн | Файл | Что посмотреть |
|
||||
|---------|------|----------------|
|
||||
| SCD1 (простое измерение) | `sql/dds/dim_airports_load.sql` | UPSERT с суррогатным ключом |
|
||||
| SCD1 с обогащением | `sql/dds/dim_airplanes_load.sql` | JOIN с `ods.seats` для `total_seats` |
|
||||
| SCD2 (версионирование) | `sql/dds/dim_routes_load.sql` | `hashdiff`, `valid_from`/`valid_to`, денормализация |
|
||||
| Факт с point-in-time JOIN | `sql/dds/fact_flight_sales_load.sql` | SCD2 lookup, защитные LEFT JOIN |
|
||||
|
||||
### DM: витрины
|
||||
|
||||
| Паттерн | Файл | Что посмотреть |
|
||||
|---------|------|----------------|
|
||||
| UPSERT с HWM по датам | `sql/dm/sales_report_load.sql` | Инкрементальная агрегация фактов |
|
||||
| Full Rebuild (AO Column) | `sql/dm/route_performance_load.sql` | TRUNCATE + INSERT, агрегация по SCD2 `route_bk` |
|
||||
| UNION ALL (dual-role dimension) | `sql/dm/airport_traffic_load.sql` | Разворот факта: departure + arrival |
|
||||
| Двухуровневая агрегация | `sql/dm/monthly_overview_load.sql` | Честный `avg_load_factor` через 2 уровня |
|
||||
| Метод затронутых ключей | `sql/dm/passenger_loyalty_load.sql` | HWM по пассажирам, пересчёт полной истории |
|
||||
|
||||
## Техническое задание
|
||||
|
||||
Основной документ: **[analyst_spec.md](analyst_spec.md)** — ТЗ от аналитика
|
||||
с описанием всех таблиц, маппингами, бизнес-правилами и подсказками.
|
||||
|
||||
## Эталон для изучения
|
||||
## Описания DAG-ов
|
||||
|
||||
Перед началом работы изучите эталонный срез (витрина `dm.sales_report`
|
||||
и вся её цепочка STG → ODS → DDS → DM):
|
||||
|
||||
- SQL-скрипты: `sql/stg/`, `sql/ods/`, `sql/dds/`, `sql/dm/`
|
||||
- DAG-файлы: `airflow/dags/`
|
||||
- [STG](../bookings_to_gp_stage.md) ·
|
||||
[ODS](../bookings_to_gp_ods.md) ·
|
||||
[DDS](../bookings_to_gp_dds.md) ·
|
||||
[DM](../bookings_to_gp_dm.md)
|
||||
- [Порядок запуска DAG-ов](../reference/dag_execution_order.md)
|
||||
|
||||
## Для менторов
|
||||
|
||||
@@ -156,18 +156,9 @@ bookings_validate
|
||||
|
||||
---
|
||||
|
||||
## 6. Бэклог: идеи для будущих итераций
|
||||
## 6. Развитие лабораторных
|
||||
|
||||
### PXF-практикум (замена STG-заданий)
|
||||
|
||||
После переноса всего STG-слоя в эталон (см. `docs/plans/2026-03-11_routes-to-reference.md`)
|
||||
студенты не практикуются в создании PXF external tables и загрузке сырых данных.
|
||||
Это важный навык для DE — нужно компенсировать отдельным заданием.
|
||||
|
||||
Варианты:
|
||||
- **Новый источник данных**: подключить CSV/JSON-файл (например, справочник городов
|
||||
или курсов валют) через PXF и загрузить в отдельную STG-таблицу
|
||||
- **Мини-задание «подключи внешний справочник»**: студент создаёт PXF external table
|
||||
для существующей таблицы источника и сравнивает результат с эталонной STG
|
||||
- **Лабораторная работа**: отдельное упражнение на создание external table +
|
||||
сравнение форматов (TEXT vs CUSTOM), профилей PXF, обработку типов
|
||||
Перед добавлением или пересмотром заданий прочитайте
|
||||
[заметку о развитии лабораторных](labs_development.md): там зафиксированы
|
||||
аудитория, решение по PXF-практикуму, предложения по обязательному циклу
|
||||
и общие критерии подготовки заданий.
|
||||
@@ -0,0 +1,200 @@
|
||||
# Развитие лабораторных работ
|
||||
|
||||
Рабочая заметка для автора программы, 22 сентября 2026 года.
|
||||
Документ хранится в ветке `solution`; оценка программы и ссылки
|
||||
на действующие задания и код относятся к учебной ветке `main`.
|
||||
Приоритет — закрыть пробелы обязательного цикла. Принятое решение о новой
|
||||
лабораторной записано отдельно от предложений, которые еще обсуждаются.
|
||||
Действующий маршрут и требования находятся в [учебных заданиях](https://git.dementev.space/ddmitry/airflow-greenplum/src/branch/main/docs/assignment/README.md)
|
||||
и [ТЗ](https://git.dementev.space/ddmitry/airflow-greenplum/src/branch/main/docs/assignment/analyst_spec.md).
|
||||
|
||||
## Аудитория и цель
|
||||
|
||||
Студент хорошо изучил PostgreSQL и прошел 16-часовой курс Greenplum от Яндекса.
|
||||
Теперь ему нужна практика применения этих знаний в работе Data Engineer.
|
||||
Стенд служит учебным пособием, похожим на небольшое хранилище компании:
|
||||
здесь есть источники, связанные слои данных, загрузки и потребители.
|
||||
|
||||
Программа должна дать опыт подключения источника, построения загрузок
|
||||
и моделей, оркестрации, проверки результата и восстановления после ошибок.
|
||||
Знакомые SQL-инструменты студент применяет к инженерным задачам и объясняет
|
||||
свои решения на данных. Разнообразие работ определяется новыми действиями
|
||||
и решениями; дополнительная витрина полезна, если она дает такой опыт.
|
||||
|
||||
Подготовка по Airflow пока не задана. Первые необходимые действия в UI
|
||||
нужно сопровождать короткой инструкцией или ссылкой на нее. Самостоятельное
|
||||
написание задач и зависимостей вводится явно в соответствующей работе.
|
||||
|
||||
## Метод и рост самостоятельности
|
||||
|
||||
Стенд учебный: его можно ломать и восстанавливать. Готовая реализация
|
||||
позволяет сначала увидеть работающую цепочку, а затем самостоятельно
|
||||
реализовать выбранный участок. Полнота эталона при первом запуске
|
||||
не ограничивает состав лабораторных.
|
||||
|
||||
Этот прием уже используется в
|
||||
[программе Clickstream](https://git.dementev.space/ddmitry/clickstream-data-platform/src/branch/main/docs/specs/2026-09-01-labs-program.md):
|
||||
при «вычитании» студент удаляет готовую реализацию и пишет свою;
|
||||
при работе «собери рядом» сравнивает свою реализацию с работающей;
|
||||
при расследовании находит и устраняет причину подготовленной ошибки.
|
||||
Можно также расширять стенд новым источником или объектом по ТЗ.
|
||||
|
||||
Границу самостоятельной работы определяет навык: в задании по SQL можно
|
||||
оставить готовый DAG, а в задании по оркестрации — готовые SQL-скрипты.
|
||||
Вычитание дает способ выделить участок для работы; самостоятельность
|
||||
обеспечивают решения, которые студент принимает и проверяет:
|
||||
|
||||
1. В первых работах даны требования к результату, близкий пример и готовая
|
||||
проверка. Студент выбирает реализацию и испытывает ее на заданных случаях.
|
||||
2. Дальше студент обосновывает способ загрузки или устройство таблицы,
|
||||
пишет собственные проверки и показывает, какие ошибки они обнаруживают.
|
||||
3. В работах на диагностику даны симптом и критерий восстановления.
|
||||
Студент сам находит причину, выбирает исправление и подтверждает результат.
|
||||
|
||||
Так помощь постепенно сокращается, а ответственность растет.
|
||||
Заранее оставлять заглушки во всех учебных участках не требуется.
|
||||
В частности, зависимость ODS от общего STG-батча объясняет
|
||||
[готовый STG в текущей программе](https://git.dementev.space/ddmitry/airflow-greenplum/src/branch/main/docs/assignment/analyst_spec.md#почему-stg-слой-уже-реализован),
|
||||
но позволяет временно разобрать его в отдельной работе.
|
||||
|
||||
## Принято: прием справочника входит в обязательную программу
|
||||
|
||||
Добавлена лабораторная **«Восстановите прием справочника из источника»**.
|
||||
Она закрывает практику подключения через PXF и загрузки полного снимка
|
||||
в STG, которые сейчас даны готовыми.
|
||||
|
||||
Предлагаемый объем: восстановить внешнюю таблицу для существующего
|
||||
справочника Postgres, написать его загрузку в STG и проверить передачу
|
||||
полей, ключей и меток батча. Студент должен различить повтор загрузки
|
||||
в том же батче и получение нового снимка. Результат должен снова
|
||||
участвовать в действующей цепочке ODS → DDS → DM.
|
||||
|
||||
В [#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).
|
||||
|
||||
## Покрытие обязательного цикла
|
||||
|
||||
Оценка основана на текущем ТЗ и коде. Наличие готового примера или главы
|
||||
для чтения учитывается отдельно от самостоятельной реализации.
|
||||
|
||||
| Участок | Что студент делает сейчас | Что стоит добавить |
|
||||
|---|---|---|
|
||||
| PXF и снимок STG | Восстанавливает прием `airports`, испытывает повтор, изменение источника и передачу в ODS | Закрыто лабораторной #9 |
|
||||
| Инкремент STG | Использует готовый отбор по дате источника | Самостоятельную загрузку новых данных и проверку границы отбора |
|
||||
| Снимки ODS | Реализует `airplanes`, `seats` и их DQ | Сохранить задания; использовать их после знакомства с приемом данных |
|
||||
| Инкремент ODS | Использует готовые загрузки | Обработку новых ключей и изменений уже полученных строк |
|
||||
| Измерения DDS | Реализует SCD1 и SCD2 | Сохранить существующий блок |
|
||||
| Факт DDS | Читает пример и запускает готовую сборку | Самостоятельную сборку факта с ключами измерений |
|
||||
| Витрины DM | Реализует четыре витрины с разными способами пересчета | Сохранить существующий блок; сначала закрыть пробелы ниже DM |
|
||||
| DDL и физическая организация | Получает готовые таблицы, типы и ключи распределения | Написать DDL одного объекта и проверить выбор распределения опытом; предложение раскрыто ниже |
|
||||
| Airflow | Запускает готовые DAG; задачи уже подключены к SQL | Самостоятельное включение загрузок и проверок в DAG |
|
||||
| DQ и повторяемость | Пишет `*_dq.sql`; для ODS проверяет повтор; пользуется `bookings_validate` | Обязательные опыты на обнаружение ошибки, идемпотентность и восстановление после сбоя |
|
||||
|
||||
## Предложения по оставшимся пробелам
|
||||
|
||||
Следующие изменения пока не утверждены как обязательные лабораторные.
|
||||
|
||||
1. **Загрузите новые и измененные данные: STG → ODS.** Кандидат —
|
||||
`bookings`: [прием в STG](https://git.dementev.space/ddmitry/airflow-greenplum/src/branch/main/sql/stg/bookings_load.sql) и
|
||||
[обновление ODS](https://git.dementev.space/ddmitry/airflow-greenplum/src/branch/main/sql/ods/bookings_load.sql). Проверить первую
|
||||
загрузку, следующую пачку, повтор и отсутствие новых данных.
|
||||
Для изменения существующего ключа подготовить новую версию в STG:
|
||||
текущий отбор источника по `book_date` сам по себе не подхватывает
|
||||
исправления старых бронирований. Различие времени события и времени
|
||||
загрузки — часть учебного результата. Разделение на две работы зависит
|
||||
от объема самостоятельного SQL.
|
||||
2. **Соберите факт продаж.** После измерений восстановить
|
||||
[загрузку факта](https://git.dementev.space/ddmitry/airflow-greenplum/src/branch/main/sql/dds/fact_flight_sales_load.sql): сохранить
|
||||
зерно `(ticket_no, flight_id)`, подобрать ключи измерений и версию
|
||||
маршрута на дату рейса, обработать обновления и повторный запуск.
|
||||
Проверить покрытие ключей, значения и отсутствие размножения строк.
|
||||
Простого совпадения количества строк недостаточно.
|
||||
3. **Организуйте загрузку в Airflow и восстановите ее после сбоя.**
|
||||
Использовать уже написанный SQL, самостоятельно связать задачи загрузки
|
||||
и DQ. На подготовленном входе вызвать сбой, определить состояние данных
|
||||
и завершить обработку повторным запуском. Генератор управляется отдельно:
|
||||
его повтор продвигает источник на новый день и меняет условия опыта.
|
||||
4. **Проверьте физическую организацию таблицы в Greenplum.** Предлагается
|
||||
включить в обязательный цикл компактный опыт: написать DDL, сравнить
|
||||
два варианта распределения одних данных, посмотреть строки по сегментам
|
||||
и планы одного запроса, обосновать выбор. Результат запроса должен
|
||||
совпадать в обоих вариантах. Наличие нескольких сегментов и достаточный
|
||||
для наблюдения объем нужно проверить при подготовке работы. Место опыта
|
||||
еще не выбрано; кандидат — задание по факту. Подробную оптимизацию
|
||||
можно развить в факультативах.
|
||||
|
||||
## Предлагаемые переходы между работами
|
||||
|
||||
Порядок обсуждения новых работ: прием справочника → инкремент STG/ODS →
|
||||
сборка факта → оркестрация и восстановление. Между ними остаются
|
||||
существующие задания на снимки ODS, измерения и витрины. Точное место
|
||||
новых работ, включая опыт с распределением, еще обсуждается.
|
||||
|
||||
Для встраивания в маршрут предлагаются следующие границы:
|
||||
|
||||
| Новая работа | Состояние перед началом | Результат для продолжения |
|
||||
|---|---|---|
|
||||
| Прием справочника | Пройден быстрый старт, работает эталонная цепочка до `sales_report` | Собственная загрузка проверена; источник восстановлен, новый общий снимок принят ODS и готов для заданий `airplanes` и `seats` |
|
||||
| Инкремент STG/ODS | Разобраны прием снимка и существующие задания ODS; подготовлены таблицы и входные пачки | Собственные загрузки дают ожидаемое текущее состояние ODS после всех проверочных случаев |
|
||||
| Сборка факта | Реализованы и проверены измерения DDS | Факт построен собственным SQL, проверены ключи и значения; на нем можно выполнять задания DM |
|
||||
| Оркестрация и восстановление | SQL выбранного участка реализован и проверен | DAG выполняет загрузку и DQ в нужном порядке; после подготовленного сбоя обработка завершается с ожидаемыми данными |
|
||||
|
||||
Готовность оценивается по нужному этапу: на исходной `main` часть таблиц
|
||||
еще является заданием, поэтому проход всех проверок будущих слоев
|
||||
не может быть условием начала первой работы. Действующие особенности
|
||||
запусков и состояния слоев описаны в [порядке выполнения DAG](https://git.dementev.space/ddmitry/airflow-greenplum/src/branch/main/docs/dag_execution_order.md).
|
||||
|
||||
Оставшиеся предложения основаны на чтении исходников; их воспроизводимость
|
||||
на стенде еще не проверялась. Состояние проверки приема `airports`
|
||||
зафиксировано отдельно в [материалах #9](pxf_airports_lab.md#проверочный-прогон).
|
||||
|
||||
## Как подготовить лабораторную
|
||||
|
||||
Эти критерии общие для обязательных работ и факультативов. Перед написанием
|
||||
инструкции определите:
|
||||
|
||||
- **Навык:** рабочая ситуация, новое действие или решение DE и его отличие
|
||||
от уже пройденных работ.
|
||||
- **Вход:** необходимые задания, состояние таблиц и загрузок, подготовка
|
||||
данных и способ воспроизвести ситуацию.
|
||||
- **Самостоятельная часть:** что студент получает готовым, что пишет
|
||||
или меняет сам и какие решения должен обосновать.
|
||||
- **Результат:** что будет работать после задания и по каким наблюдениям
|
||||
студент сможет это подтвердить.
|
||||
- **Проверочные случаи:** конкретные данные или события, ожидаемый результат
|
||||
каждого случая и способ обнаружить неверное решение.
|
||||
- **Выход:** что сохраняется для следующей работы и как получить нужное
|
||||
состояние стенда после опыта или перерыва.
|
||||
|
||||
Для инкремента получение новых строк из источника и применение уже
|
||||
полученных изменений в ODS проверяются отдельно. Для собственной DQ-проверки
|
||||
нужны отказ на известной ошибке и успех после исправления. Сверка с готовой
|
||||
реализацией подтверждает совпадение поведения; проверки по источнику
|
||||
подтверждают полноту и бизнес-правила.
|
||||
|
||||
Полная пересборка стенда допустима. Адресное восстановление требуется там,
|
||||
где оно само является предметом работы. При пересборке инструкция должна
|
||||
объяснять, как снова применить сохраненные файлы решения и получить
|
||||
состояние для продолжения.
|
||||
|
||||
Лабораторная готова к выдаче, когда все ее проверочные случаи пройдены
|
||||
на стенде из описанного начального состояния, проверки обнаружили
|
||||
предусмотренные ошибки, а заявленный выход позволяет продолжить маршрут.
|
||||
Команды и результаты проверки сохраняются вместе с изменением задания.
|
||||
Текстовая вычитка и проверка ссылок подтверждают только качество документа.
|
||||
|
||||
## Основа для будущих факультативов
|
||||
|
||||
Факультативы расширяют опыт за пределами обязательного набора. При генерации
|
||||
кандидатов используйте общие критерии выше и указывайте, после каких
|
||||
обязательных работ доступен каждый сюжет.
|
||||
|
||||
Направления для последующей генерации: изменения схемы источника,
|
||||
опоздавшие данные, история и пересчеты, углубленная оптимизация запросов
|
||||
Greenplum, свежесть данных, разграничение доступа, новые способы приема.
|
||||
Это исходные темы для отбора, а не утвержденный список заданий.
|
||||
@@ -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` заполняет дополнительные
|
||||
таблицы и не подтверждает проходимость первой работы с учебными заглушками.
|
||||
@@ -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
|
||||
```
|
||||
@@ -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())
|
||||
@@ -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")
|
||||
@@ -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")
|
||||
|
||||
Reference in new issue
Block a user