diff --git a/plans/stg_layer_implementation_plan.md b/plans/stg_layer_implementation_plan.md new file mode 100644 index 0000000..c149f6c --- /dev/null +++ b/plans/stg_layer_implementation_plan.md @@ -0,0 +1,730 @@ +# План реализации STG слоя целиком + +> **Статус:** План готов к реализации +> **Дата:** 2026-01-17 +> **Автор:** Architect Mode + +## Обзор задачи + +Согласно [`docs/internal/db_schema.md`](../docs/internal/db_schema.md:31), STG слой реализован частично (2 из 9 таблиц: bookings, tickets). Необходимо реализовать оставшиеся 7 таблиц. + +## Стратегия загрузки данных + +| Тип таблиц | Стратегия | Обоснование | +|-----------|-----------|-------------| +| Справочники (airports, airplanes, routes, seats) | **Full load** | Маленький объём (<10K строк), простота реализации | +| Транзакции (flights, segments, boarding_passes) | **Инкремент по дате** | Больший объём, необходимость отслеживания изменений | + +### Оценка размера справочников + +| Справочник | Примерный размер | Оценка | +|-------------|------------------|---------| +| **airports** | ~700-800 аэропортов | **Маленький** | +| **airplanes** | ~10 моделей самолётов | **Крошечный** | +| **seats** | ~1700-2000 записей | **Маленький** | +| **routes** | Ожидается несколько тысяч | **Маленький/Средний** | + +**Вывод:** Все справочники очень маленькие (до нескольких тысяч строк). Даже если routes будет 5000-10000 строк - это всё равно минимальный объём для Greenplum. + +## Список таблиц для реализации + +| Таблица источника | Таблица STG | Тип данных | Стратегия загрузки | Опорное поле для инкремента | +|-------------------|-------------|------------|-------------------|---------------------------| +| `bookings.airports_data` | `stg.airports` | Справочник | Full | - | +| `bookings.airplanes_data` | `stg.airplanes` | Справочник | Full | - | +| `bookings.routes` | `stg.routes` | Справочник | Full | - | +| `bookings.seats` | `stg.seats` | Справочник | Full | - | +| `bookings.flights` | `stg.flights` | Транзакции | Инкремент | `scheduled_departure` | +| `bookings.segments` | `stg.segments` | Транзакции | Инкремент | `book_date` (через tickets) | +| `bookings.boarding_passes` | `stg.boarding_passes` | Транзакции | Инкремент | `book_date` (через tickets) | + +## Паттерн реализации (на основе bookings/tickets) + +Для каждой таблицы создаются 3 файла: + +1. **`sql/stg/{table}_ddl.sql`** - DDL для внешней и внутренней таблиц +2. **`sql/stg/{table}_load.sql`** - Загрузка (full или инкремент) +3. **`sql/stg/{table}_dq.sql`** - Проверки качества данных + +### Общая структура DDL файла + +```sql +-- DDL для слоя STG по таблице {table}. +-- Используется как из общего скрипта ddl_gp.sql (через \i), +-- так и может выполняться отдельно при изменении схемы. + +-- Схема stg для сырого слоя DWH. +CREATE SCHEMA IF NOT EXISTS stg; + +-- Внешняя таблица в схеме stg для чтения данных из bookings.{table} через PXF. +DROP EXTERNAL TABLE IF EXISTS stg.{table}_ext; +CREATE EXTERNAL TABLE stg.{table}_ext ( + -- поля из источника +) +LOCATION ('pxf://bookings.{table}?PROFILE=JDBC&SERVER=bookings-db') +FORMAT 'CUSTOM' (formatter='pxfwritable_import'); + +-- Внутренняя таблица stg.{table} — сырой слой, все бизнес-колонки как TEXT. +CREATE TABLE IF NOT EXISTS stg.{table} ( + -- бизнес-колонки как TEXT + src_created_at_ts TIMESTAMP, + load_dttm TIMESTAMP NOT NULL DEFAULT now(), + batch_id TEXT NOT NULL +) +WITH (appendonly=true, orientation=row, compresstype=zlib, compresslevel=1) +DISTRIBUTED BY ({distribution_key}); +``` + +### Общая структура LOAD файла (Full load для справочников) + +```sql +-- Загрузка всех строк из stg.{table}_ext в stg.{table}. +-- Используем batch_id для отслеживания загрузки. + +INSERT INTO stg.{table} ( + -- бизнес-колонки + src_created_at_ts, + load_dttm, + batch_id +) +SELECT + ext.{field}::text, + now()::timestamp, + '{{ run_id }}'::text +FROM stg.{table}_ext AS ext +WHERE NOT EXISTS ( + -- Защита от дублей в рамках одного batch_id + SELECT 1 + FROM stg.{table} AS t + WHERE t.batch_id = '{{ run_id }}'::text + AND t.{pk} = ext.{pk}::text +); + +-- Обновляем статистику для оптимизатора Greenplum +ANALYZE stg.{table}; +``` + +### Общая структура LOAD файла (Инкремент для транзакций) + +```sql +-- Загрузка инкремента из stg.{table}_ext в stg.{table}. +-- Окно инкремента определяется по src_created_at_ts: +-- берём строки, где {increment_field} больше максимального src_created_at_ts +-- среди "старых" батчей; верхняя граница по дате не используется. + +-- CTE для определения максимальной даты загрузки предыдущего батча +WITH max_batch_ts AS ( + SELECT COALESCE(MAX(src_created_at_ts), TIMESTAMP '1900-01-01 00:00:00') AS max_ts + FROM stg.{table} + WHERE batch_id <> '{{ run_id }}'::text + OR batch_id IS NULL +) +INSERT INTO stg.{table} ( + -- бизнес-колонки + src_created_at_ts, + load_dttm, + batch_id +) +SELECT + ext.{field}::text, + ext.{increment_field}::timestamp, + now(), + '{{ run_id }}'::text +FROM stg.{table}_ext AS ext +CROSS JOIN max_batch_ts AS mb +WHERE ext.{increment_field} > mb.max_ts +AND NOT EXISTS ( + SELECT 1 + FROM stg.{table} AS t + WHERE t.batch_id = '{{ run_id }}'::text + AND t.{pk} = ext.{pk}::text +); + +-- Обновляем статистику для оптимизатора Greenplum +ANALYZE stg.{table}; +``` + +### Общая структура DQ файла + +```sql +-- Проверки качества данных для {table} + +DO $$ +DECLARE + v_batch_id TEXT := '{{ run_id }}'::text; + v_prev_ts TIMESTAMP; + v_src_count BIGINT; + v_stg_count BIGINT; + v_dup_count BIGINT; + v_null_count BIGINT; + -- другие переменные для специфических проверок +BEGIN + -- Опорная метка: максимум src_created_at_ts среди предыдущих батчей + SELECT max(src_created_at_ts) + INTO v_prev_ts + FROM stg.{table} + WHERE batch_id <> v_batch_id + OR batch_id IS NULL; + + -- Источник: считаем строки во внешней таблице, которые вошли в окно инкремента + SELECT COUNT(*) + INTO v_src_count + FROM stg.{table}_ext + WHERE {increment_field} > COALESCE(v_prev_ts, TIMESTAMP '1900-01-01 00:00:00'); + + IF v_src_count = 0 THEN + RAISE EXCEPTION + 'В источнике {table}_ext нет строк для окна инкремента.'; + END IF; + + -- Считаем строки, реально вставленные в stg.{table} в этом батче + SELECT COUNT(*) + INTO v_stg_count + FROM stg.{table} + WHERE batch_id = v_batch_id; + + IF v_src_count <> v_stg_count THEN + RAISE EXCEPTION + 'DQ FAILED: несовпадение количества строк. Источник: %, STG: %', + v_src_count, + v_stg_count; + END IF; + + -- Проверка на дубликаты первичного ключа + SELECT COUNT(*) - COUNT(DISTINCT {pk}) + INTO v_dup_count + FROM stg.{table} AS t + WHERE t.batch_id = v_batch_id; + + IF v_dup_count <> 0 THEN + RAISE EXCEPTION + 'DQ FAILED: найдены дубликаты {pk} (batch_id=%): %', + v_batch_id, + v_dup_count; + END IF; + + -- Проверка обязательных полей + SELECT COUNT(*) + INTO v_null_count + FROM stg.{table} AS t + WHERE t.batch_id = v_batch_id + AND (t.{required_field} IS NULL OR t.{required_field} = ''); + + IF v_null_count <> 0 THEN + RAISE EXCEPTION + 'DQ FAILED: найдены строки с NULL в обязательных полях (batch_id=%): %', + v_batch_id, + v_null_count; + END IF; + + RAISE NOTICE + 'DQ PASSED: {table} ок (batch_id=%): source=% stg=%', + v_batch_id, + v_src_count, + v_stg_count; +END $$; +``` + +## Детали реализации по таблицам + +### 1. airports (справочник, full load) + +**Внешняя таблица**: `stg.airports_ext` +- Поля: `airport_code`, `airport_name` (JSONB), `city` (JSONB), `country` (JSONB), `coordinates`, `timezone` +- PXF: `pxf://bookings.airports_data?PROFILE=JDBC&SERVER=bookings-db` + +**Внутренняя таблица**: `stg.airports` +- Бизнес-колонки как TEXT: + - `airport_code TEXT` + - `airport_name TEXT` + - `city TEXT` + - `country TEXT` + - `coordinates TEXT` + - `timezone TEXT` +- Тех.колонки: `src_created_at_ts`, `load_dttm`, `batch_id` +- Распределение: `DISTRIBUTED BY (airport_code)` +- Обоснование: `airport_code` — это уникальный идентификатор аэропорта + +**Загрузка**: Full (все строки при каждом запуске) + +**DQ проверки**: +- Count между источником и STG +- Дубликаты `airport_code` +- NULL обязательных полей (airport_code, airport_name, city, timezone) + +### 2. airplanes (справочник, full load) + +**Внешняя таблица**: `stg.airplanes_ext` +- Поля: `airplane_code`, `model` (JSONB), `range`, `speed` +- PXF: `pxf://bookings.airplanes_data?PROFILE=JDBC&SERVER=bookings-db` + +**Внутренняя таблица**: `stg.airplanes` +- Бизнес-колонки как TEXT: + - `airplane_code TEXT` + - `model TEXT` + - `range TEXT` + - `speed TEXT` +- Тех.колонки: `src_created_at_ts`, `load_dttm`, `batch_id` +- Распределение: `DISTRIBUTED BY (airplane_code)` +- Обоснование: `airplane_code` — это уникальный идентификатор самолёта + +**Загрузка**: Full + +**DQ проверки**: +- Count между источником и STG +- Дубликаты `airplane_code` +- NULL обязательных полей (airplane_code, model) + +### 3. routes (справочник, full load) + +**Внешняя таблица**: `stg.routes_ext` +- Поля: `route_no`, `validity` (tstzrange), `departure_airport`, `arrival_airport`, `airplane_code`, `days_of_week` (int[]), `scheduled_time`, `duration` +- PXF: `pxf://bookings.routes?PROFILE=JDBC&SERVER=bookings-db` + +**Внутренняя таблица**: `stg.routes` +- Бизнес-колонки как TEXT: + - `route_no TEXT` + - `validity TEXT` + - `departure_airport TEXT` + - `arrival_airport TEXT` + - `airplane_code TEXT` + - `days_of_week TEXT` + - `scheduled_time TEXT` + - `duration TEXT` +- Тех.колонки: `src_created_at_ts`, `load_dttm`, `batch_id` +- Распределение: `DISTRIBUTED BY (route_no)` +- Обоснование: `route_no` — это уникальный идентификатор маршрута + +**Загрузка**: Full + +**DQ проверки**: +- Count между источником и STG +- Дубликаты `route_no` +- NULL обязательных полей (route_no, departure_airport, arrival_airport, airplane_code) +- Ссылочная целостность на airports (departure_airport, arrival_airport) +- Ссылочная целостность на airplanes (airplane_code) + +### 4. seats (справочник, full load) + +**Внешняя таблица**: `stg.seats_ext` +- Поля: `airplane_code`, `seat_no`, `fare_conditions` +- PXF: `pxf://bookings.seats?PROFILE=JDBC&SERVER=bookings-db` + +**Внутренняя таблица**: `stg.seats` +- Бизнес-колонки как TEXT: + - `airplane_code TEXT` + - `seat_no TEXT` + - `fare_conditions TEXT` +- Тех.колонки: `src_created_at_ts`, `load_dttm`, `batch_id` +- Распределение: `DISTRIBUTED BY (airplane_code)` +- Обоснование: co-location с airplanes для оптимизации JOIN + +**Загрузка**: Full + +**DQ проверки**: +- Count между источником и STG +- Дубликаты `(airplane_code, seat_no)` +- NULL обязательных полей (airplane_code, seat_no, fare_conditions) +- Ссылочная целостность на airplanes (airplane_code) + +### 5. flights (транзакции, инкремент) + +**Внешняя таблица**: `stg.flights_ext` +- Поля: `flight_id`, `route_no`, `status`, `scheduled_departure`, `scheduled_arrival`, `actual_departure`, `actual_arrival` +- PXF: `pxf://bookings.flights?PROFILE=JDBC&SERVER=bookings-db` + +**Внутренняя таблица**: `stg.flights` +- Бизнес-колонки как TEXT: + - `flight_id TEXT` + - `route_no TEXT` + - `status TEXT` + - `scheduled_departure TEXT` + - `scheduled_arrival TEXT` + - `actual_departure TEXT` + - `actual_arrival TEXT` +- Тех.колонки: `src_created_at_ts`, `load_dttm`, `batch_id` +- `src_created_at_ts` = `scheduled_departure` +- Распределение: `DISTRIBUTED BY (flight_id)` +- Обоснование: `flight_id` — это уникальный идентификатор рейса + +**Загрузка**: Инкремент по `scheduled_departure` + +**DQ проверки**: +- Count между источником и STG +- Дубликаты `flight_id` +- NULL обязательных полей (flight_id, route_no, status, scheduled_departure) +- Ссылочная целостность на routes (route_no) + +### 6. segments (транзакции, инкремент) + +**Внешняя таблица**: `stg.segments_ext` +- Поля: `ticket_no`, `flight_id`, `fare_conditions`, `price` +- PXF: `pxf://bookings.segments?PROFILE=JDBC&SERVER=bookings-db` + +**Внутренняя таблица**: `stg.segments` +- Бизнес-колонки как TEXT: + - `ticket_no TEXT` + - `flight_id TEXT` + - `fare_conditions TEXT` + - `price TEXT` +- Тех.колонки: `src_created_at_ts`, `load_dttm`, `batch_id` +- `src_created_at_ts` = берётся из `bookings.book_date` через JOIN с tickets +- Распределение: `DISTRIBUTED BY (ticket_no)` +- Обоснование: co-location с tickets для оптимизации JOIN + +**Загрузка**: Инкремент по `book_date` (как в tickets) + +**DQ проверки**: +- Count между источником и STG +- Дубликаты `(ticket_no, flight_id)` +- NULL обязательных полей (ticket_no, flight_id, fare_conditions, price) +- Ссылочная целостность на tickets (ticket_no) +- Ссылочная целостность на flights (flight_id) + +### 7. boarding_passes (транзакции, инкремент) + +**Внешняя таблица**: `stg.boarding_passes_ext` +- Поля: `ticket_no`, `flight_id`, `seat_no`, `boarding_no`, `boarding_time` +- PXF: `pxf://bookings.boarding_passes?PROFILE=JDBC&SERVER=bookings-db` + +**Внутренняя таблица**: `stg.boarding_passes` +- Бизнес-колонки как TEXT: + - `ticket_no TEXT` + - `flight_id TEXT` + - `seat_no TEXT` + - `boarding_no TEXT` + - `boarding_time TEXT` +- Тех.колонки: `src_created_at_ts`, `load_dttm`, `batch_id` +- `src_created_at_ts` = берётся из `bookings.book_date` через JOIN с tickets +- Распределение: `DISTRIBUTED BY (ticket_no)` +- Обоснование: co-location с tickets/segments для оптимизации JOIN + +**Загрузка**: Инкремент по `book_date` (как в tickets) + +**DQ проверки**: +- Count между источником и STG +- Дубликаты `(ticket_no, flight_id)` +- NULL обязательных полей (ticket_no, flight_id) +- Ссылочная целостность на tickets (ticket_no) +- Ссылочная целостность на segments (ticket_no, flight_id) + +## Обновление существующих DAG + +### `airflow/dags/bookings_stg_ddl.py` + +Добавить задачи для создания DDL новых таблиц: + +```python +apply_stg_airports_ddl = PostgresOperator( + task_id="apply_stg_airports_ddl", + postgres_conn_id=GREENPLUM_CONN_ID, + sql="stg/airports_ddl.sql", +) + +apply_stg_airplanes_ddl = PostgresOperator( + task_id="apply_stg_airplanes_ddl", + postgres_conn_id=GREENPLUM_CONN_ID, + sql="stg/airplanes_ddl.sql", +) + +apply_stg_routes_ddl = PostgresOperator( + task_id="apply_stg_routes_ddl", + postgres_conn_id=GREENPLUM_CONN_ID, + sql="stg/routes_ddl.sql", +) + +apply_stg_seats_ddl = PostgresOperator( + task_id="apply_stg_seats_ddl", + postgres_conn_id=GREENPLUM_CONN_ID, + sql="stg/seats_ddl.sql", +) + +apply_stg_flights_ddl = PostgresOperator( + task_id="apply_stg_flights_ddl", + postgres_conn_id=GREENPLUM_CONN_ID, + sql="stg/flights_ddl.sql", +) + +apply_stg_segments_ddl = PostgresOperator( + task_id="apply_stg_segments_ddl", + postgres_conn_id=GREENPLUM_CONN_ID, + sql="stg/segments_ddl.sql", +) + +apply_stg_boarding_passes_ddl = PostgresOperator( + task_id="apply_stg_boarding_passes_ddl", + postgres_conn_id=GREENPLUM_CONN_ID, + sql="stg/boardings_ddl.sql", +) +``` + +Зависимости: +- Сначала создаются справочники (airports, airplanes, routes, seats) +- Затем транзакционные таблицы (flights, segments, boarding_passes) + +### `airflow/dags/bookings_to_gp_stage.py` + +Добавить задачи для загрузки новых таблиц: + +```python +# Загрузка справочников (full load) +load_airports_to_stg = PostgresOperator( + task_id="load_airports_to_stg", + postgres_conn_id=GREENPLUM_CONN_ID, + sql="stg/airports_load.sql", +) + +check_airports_dq = PostgresOperator( + task_id="check_airports_dq", + postgres_conn_id=GREENPLUM_CONN_ID, + sql="stg/airports_dq.sql", +) + +load_airplanes_to_stg = PostgresOperator( + task_id="load_airplanes_to_stg", + postgres_conn_id=GREENPLUM_CONN_ID, + sql="stg/airplanes_load.sql", +) + +check_airplanes_dq = PostgresOperator( + task_id="check_airplanes_dq", + postgres_conn_id=GREENPLUM_CONN_ID, + sql="stg/airplanes_dq.sql", +) + +load_routes_to_stg = PostgresOperator( + task_id="load_routes_to_stg", + postgres_conn_id=GREENPLUM_CONN_ID, + sql="stg/routes_load.sql", +) + +check_routes_dq = PostgresOperator( + task_id="check_routes_dq", + postgres_conn_id=GREENPLUM_CONN_ID, + sql="stg/routes_dq.sql", +) + +load_seats_to_stg = PostgresOperator( + task_id="load_seats_to_stg", + postgres_conn_id=GREENPLUM_CONN_ID, + sql="stg/seats_load.sql", +) + +check_seats_dq = PostgresOperator( + task_id="check_seats_dq", + postgres_conn_id=GREENPLUM_CONN_ID, + sql="stg/seats_dq.sql", +) + +# Загрузка транзакций (инкремент) +load_flights_to_stg = PostgresOperator( + task_id="load_flights_to_stg", + postgres_conn_id=GREENPLUM_CONN_ID, + sql="stg/flights_load.sql", +) + +check_flights_dq = PostgresOperator( + task_id="check_flights_dq", + postgres_conn_id=GREENPLUM_CONN_ID, + sql="stg/flights_dq.sql", +) + +load_segments_to_stg = PostgresOperator( + task_id="load_segments_to_stg", + postgres_conn_id=GREENPLUM_CONN_ID, + sql="stg/segments_load.sql", +) + +check_segments_dq = PostgresOperator( + task_id="check_segments_dq", + postgres_conn_id=GREENPLUM_CONN_ID, + sql="stg/segments_dq.sql", +) + +load_boarding_passes_to_stg = PostgresOperator( + task_id="load_boarding_passes_to_stg", + postgres_conn_id=GREENPLUM_CONN_ID, + sql="stg/boardings_load.sql", +) + +check_boarding_passes_dq = PostgresOperator( + task_id="check_boarding_passes_dq", + postgres_conn_id=GREENPLUM_CONN_ID, + sql="stg/boardings_dq.sql", +) +``` + +Зависимости: +- Сначала загружаются и проверяются bookings и tickets (уже есть) +- Затем загружаются справочники (airports, airplanes, routes, seats) +- Затем загружаются транзакции (flights, segments, boarding_passes) +- В конце финальный лог + +### Обновление `sql/ddl_gp.sql` + +Добавить подключение новых DDL файлов: + +```sql +-- DDL для слоя stg по таблицам bookings и tickets вынесены в отдельные файлы. +-- Здесь подключаем их через psql \i, чтобы сохранить единый входной скрипт. +\i stg/bookings_ddl.sql +\i stg/tickets_ddl.sql + +-- DDL для новых таблиц STG слоя +\i stg/airports_ddl.sql +\i stg/airplanes_ddl.sql +\i stg/routes_ddl.sql +\i stg/seats_ddl.sql +\i stg/flights_ddl.sql +\i stg/segments_ddl.sql +\i stg/boardings_ddl.sql +``` + +### Добавление тестов + +Обновить `tests/test_dags_smoke.py` для проверки структуры обновлённых DAG: + +```python +def test_bookings_stg_ddl_dag_structure(): + dag = _load_dag("airflow.dags.bookings_stg_ddl") + + expected_tasks = { + "apply_stg_bookings_ddl", + "apply_stg_tickets_ddl", + "apply_stg_airports_ddl", + "apply_stg_airplanes_ddl", + "apply_stg_routes_ddl", + "apply_stg_seats_ddl", + "apply_stg_flights_ddl", + "apply_stg_segments_ddl", + "apply_stg_boarding_passes_ddl", + } + assert expected_tasks.issubset(dag.task_dict.keys()) + + # Проверка линейных зависимостей + # ... (проверка зависимостей между задачами) +``` + +```python +def test_bookings_to_gp_stage_dag_structure(): + dag = _load_dag("airflow.dags.bookings_to_gp_stage") + + expected_tasks = { + "generate_bookings_day", + "load_bookings_to_stg", + "check_row_counts", + "load_tickets_to_stg", + "check_tickets_dq", + "load_airports_to_stg", + "check_airports_dq", + "load_airplanes_to_stg", + "check_airplanes_dq", + "load_routes_to_stg", + "check_routes_dq", + "load_seats_to_stg", + "check_seats_dq", + "load_flights_to_stg", + "check_flights_dq", + "load_segments_to_stg", + "check_segments_dq", + "load_boarding_passes_to_stg", + "check_boarding_passes_dq", + "finish_summary", + } + assert expected_tasks.issubset(dag.task_dict.keys()) + + # Проверка линейных зависимостей + # ... (проверка зависимостей между задачами) +``` + +### Обновление документации + +Обновить статус в [`docs/internal/db_schema.md`](../docs/internal/db_schema.md:31) с "2 из 9" на "9 из 9". + +Добавить описание новых таблиц в документацию. + +## Диаграмма потока данных STG слоя + +```mermaid +graph TB + subgraph Source[Source: bookings-db] + B1[airports_data] + B2[airplanes_data] + B3[routes] + B4[seats] + B5[flights] + B6[segments] + B7[boarding_passes] + end + + subgraph STG[STG Layer: Greenplum] + S1[stg.airports] + S2[stg.airplanes] + S3[stg.routes] + S4[stg.seats] + S5[stg.flights] + S6[stg.segments] + S7[stg.boarding_passes] + end + + B1 --> S1 + B2 --> S2 + B3 --> S3 + B4 --> S4 + B5 --> S5 + B6 --> S6 + B7 --> S7 +``` + +## Чек-лист реализации + +- [ ] Создать файлы DDL для новых таблиц (7 файлов) + - [ ] `sql/stg/airports_ddl.sql` + - [ ] `sql/stg/airplanes_ddl.sql` + - [ ] `sql/stg/routes_ddl.sql` + - [ ] `sql/stg/seats_ddl.sql` + - [ ] `sql/stg/flights_ddl.sql` + - [ ] `sql/stg/segments_ddl.sql` + - [ ] `sql/stg/boardings_ddl.sql` +- [ ] Создать файлы LOAD для новых таблиц (7 файлов) + - [ ] `sql/stg/airports_load.sql` + - [ ] `sql/stg/airplanes_load.sql` + - [ ] `sql/stg/routes_load.sql` + - [ ] `sql/stg/seats_load.sql` + - [ ] `sql/stg/flights_load.sql` + - [ ] `sql/stg/segments_load.sql` + - [ ] `sql/stg/boardings_load.sql` +- [ ] Создать файлы DQ для новых таблиц (7 файлов) + - [ ] `sql/stg/airports_dq.sql` + - [ ] `sql/stg/airplanes_dq.sql` + - [ ] `sql/stg/routes_dq.sql` + - [ ] `sql/stg/seats_dq.sql` + - [ ] `sql/stg/flights_dq.sql` + - [ ] `sql/stg/segments_dq.sql` + - [ ] `sql/stg/boardings_dq.sql` +- [ ] Обновить DAG `bookings_stg_ddl.py` +- [ ] Обновить DAG `bookings_to_gp_stage.py` +- [ ] Обновить `sql/ddl_gp.sql` +- [ ] Добавить тесты для новых DAG в `tests/test_dags_smoke.py` +- [ ] Обновить документацию `docs/internal/db_schema.md` +- [ ] Провести тестирование реализации + +## Примечания для реализации + +1. **Именование файлов**: Использовать `{table}_ddl.sql`, `{table}_load.sql`, `{table}_dq.sql` +2. **Ключи распределения**: Выбирать ключи с высокой кардинальностью для равномерного распределения +3. **Co-location**: Использовать одинаковые ключи распределения для связанных таблиц (tickets, segments, boarding_passes по ticket_no) +4. **Комментарии**: Добавлять русскоязычные комментарии в SQL-файлы для студентов +5. **DQ проверки**: Все проверки должны падать с `RAISE EXCEPTION` при ошибке +6. **Batch ID**: Использовать `{{ run_id }}` для идентификации батча +7. **Защита от дублей**: Использовать `NOT EXISTS` для предотвращения дублирования в рамках одного batch_id + +## Связанные документы + +- [`docs/internal/db_schema.md`](../docs/internal/db_schema.md) - Общая схема Б DWH +- [`docs/internal/bookings_stg_design.md`](../docs/internal/bookings_stg_design.md) - Детальный дизайн STG для bookings +- [`sql/stg/bookings_ddl.sql`](../sql/stg/bookings_ddl.sql) - Образец DDL +- [`sql/stg/bookings_load.sql`](../sql/stg/bookings_load.sql) - Образец LOAD +- [`sql/stg/bookings_dq.sql`](../sql/stg/bookings_dq.sql) - Образец DQ +- [`airflow/dags/bookings_stg_ddl.py`](../airflow/dags/bookings_stg_ddl.py) - Образец DAG DDL +- [`airflow/dags/bookings_to_gp_stage.py`](../airflow/dags/bookings_to_gp_stage.py) - Образец DAG загрузки