airflow-greenplum/docs/internal/bookings_ods_design.md

ODS Layer: эталонный учебный план реализации (v2)

Контекст

STG-слой уже реализован как учебный эталон:

• данные из bookings-db читаются через PXF;
• в STG бизнес-колонки хранятся как TEXT;
• загрузка и DQ работают батчами (batch_id = {{ run_id }}).

Этот документ фиксирует простую и каноничную реализацию ODS для менти.

---

1) Что считаем эталоном для ODS

1.1. Роль ODS в этом стенде

ODS в учебном проекте — это:

• типизированные и очищенные данные;
• одна актуальная запись на бизнес-ключ;
• удобный слой для последующей сборки DDS/DM.

1.2. Что делаем, что не делаем

Делаем в ODS:

• приведение типов (TEXT -> TIMESTAMPTZ/NUMERIC/INT/BOOLEAN/...);
• дедупликацию внутри батча;
• UPSERT (SCD Type 1): обновляем текущую запись при изменении, вставляем новые.

Не делаем в ODS (в базовом эталоне):

• SCD Type 2 с периодами действия;
• сложную обработку late-arriving/backdated событий;
• отдельный DQ-слой с хранением результатов.

1.3. Где хранится история изменений

• История «как приходили данные» уже сохраняется в STG (append + batch_id).
• Историзацию измерений (SCD2) показываем позже в DDS (как в учебной статье dwh-modeling).

Итог: ODS = текущий слой (current state), простой и понятный.

---

2) Нейминг служебных полей (консистентно с de-roadmap)

Источник правил: docs/internal/naming_conventions.md.

В ODS используем такие техполя:

• _load_id TEXT NOT NULL — идентификатор загрузки (берём stg_batch_id);
• _load_ts TIMESTAMP NOT NULL DEFAULT now() — время загрузки в ODS;
• event_ts TIMESTAMP — время события из источника (если у сущности оно есть).

2.1. Маппинг из текущего STG

• stg.batch_id -> ods._load_id
• stg.load_dttm не переносим 1:1; в ODS пишем собственный ods._load_ts = now()
• stg.src_created_at_ts -> ods.event_ts (для транзакционных таблиц)

2.2. Почему так

• нейминг совпадает с учебной статьёй (_load_id, _load_ts);
• студентам проще переносить паттерн между проектами;
• разделяем «когда событие произошло» (event_ts) и «когда загрузили в слой» (_load_ts).

---

3) Гранулярность и бизнес-ключи ODS

Таблица	Зерно	Бизнес-ключ
ods.airports	1 строка = аэропорт	airport_code
ods.airplanes	1 строка = самолёт	airplane_code
ods.routes	1 строка = версия маршрута	(route_no, validity)
ods.seats	1 строка = место в самолёте	(airplane_code, seat_no)
ods.bookings	1 строка = бронирование	book_ref
ods.tickets	1 строка = билет	ticket_no
ods.flights	1 строка = рейс	flight_id
ods.segments	1 строка = сегмент билета	(ticket_no, flight_id)
ods.boarding_passes	1 строка = посадочный на сегмент	(ticket_no, flight_id)

Критично для эталона:

• routes — составной ключ (route_no, validity);
• boarding_passes — составной ключ (ticket_no, flight_id).

---

4) Схема ODS-таблиц (v1, без SCD2)

Ниже — учебный минимум колонок. При необходимости можно добавлять бизнес-атрибуты без изменения паттерна загрузки.

4.1. Справочники

ods.airports

airport_code  TEXT NOT NULL
airport_name  TEXT NOT NULL
city          TEXT NOT NULL
country       TEXT NOT NULL
coordinates   TEXT
timezone      TEXT NOT NULL
_load_id      TEXT NOT NULL
_load_ts      TIMESTAMP NOT NULL DEFAULT now()
DISTRIBUTED BY (airport_code)

ods.airplanes

airplane_code TEXT NOT NULL
model         TEXT NOT NULL
range_km      INTEGER
speed_kmh     INTEGER
_load_id      TEXT NOT NULL
_load_ts      TIMESTAMP NOT NULL DEFAULT now()
DISTRIBUTED BY (airplane_code)

ods.routes

route_no            TEXT NOT NULL
validity            TEXT NOT NULL
departure_airport   TEXT NOT NULL
arrival_airport     TEXT NOT NULL
airplane_code       TEXT NOT NULL
days_of_week        TEXT
scheduled_departure_time TIME
scheduled_duration  INTERVAL
_load_id            TEXT NOT NULL
_load_ts            TIMESTAMP NOT NULL DEFAULT now()
DISTRIBUTED BY (route_no)

ods.seats

airplane_code   TEXT NOT NULL
seat_no         TEXT NOT NULL
fare_conditions TEXT NOT NULL
_load_id        TEXT NOT NULL
_load_ts        TIMESTAMP NOT NULL DEFAULT now()
DISTRIBUTED BY (airplane_code)

4.2. Транзакционные

ods.bookings

book_ref      TEXT NOT NULL
book_date     TIMESTAMP WITH TIME ZONE NOT NULL
total_amount  NUMERIC(10,2) NOT NULL
event_ts      TIMESTAMP
_load_id      TEXT NOT NULL
_load_ts      TIMESTAMP NOT NULL DEFAULT now()
DISTRIBUTED BY (book_ref)

ods.tickets

ticket_no       TEXT NOT NULL
book_ref        TEXT NOT NULL
passenger_id    TEXT NOT NULL
passenger_name  TEXT NOT NULL
is_outbound     BOOLEAN
event_ts        TIMESTAMP
_load_id        TEXT NOT NULL
_load_ts        TIMESTAMP NOT NULL DEFAULT now()
DISTRIBUTED BY (book_ref)

ods.flights

flight_id            INTEGER NOT NULL
route_no             TEXT NOT NULL
status               TEXT NOT NULL
scheduled_departure  TIMESTAMP WITH TIME ZONE
scheduled_arrival    TIMESTAMP WITH TIME ZONE
actual_departure     TIMESTAMP WITH TIME ZONE
actual_arrival       TIMESTAMP WITH TIME ZONE
event_ts             TIMESTAMP
_load_id             TEXT NOT NULL
_load_ts             TIMESTAMP NOT NULL DEFAULT now()
DISTRIBUTED BY (flight_id)

ods.segments

ticket_no        TEXT NOT NULL
flight_id        INTEGER NOT NULL
fare_conditions  TEXT NOT NULL
segment_amount   NUMERIC(10,2)
event_ts         TIMESTAMP
_load_id         TEXT NOT NULL
_load_ts         TIMESTAMP NOT NULL DEFAULT now()
DISTRIBUTED BY (ticket_no)

ods.boarding_passes

ticket_no      TEXT NOT NULL
flight_id      INTEGER NOT NULL
seat_no        TEXT NOT NULL
boarding_no    INTEGER
boarding_time  TIMESTAMP WITH TIME ZONE
event_ts       TIMESTAMP
_load_id       TEXT NOT NULL
_load_ts       TIMESTAMP NOT NULL DEFAULT now()
DISTRIBUTED BY (ticket_no)

Примечание: в учебном варианте не опираемся на физические PK/FK-constraint в Greenplum, а проверяем целостность через DQ-скрипты.

---

5) Контракт батча для ODS

Чтобы ODS был воспроизводимым, в каждом запуске используем один фиксированный stg_batch_id.

5.1. Источник stg_batch_id

В bookings_to_gp_ods:

• принимаем stg_batch_id из dag_run.conf;
• если не передан — берём последний из stg.bookings;
• логируем, какой stg_batch_id выбран.

5.2. Как применяем

Во всех sql/ods/*_load.sql:

• читаем STG только с WHERE batch_id = :stg_batch_id;
• пишем в ODS _load_id = :stg_batch_id, _load_ts = now().

Это простой и понятный паттерн: один запуск ODS = один снимок STG-батча.

---

6) SQL-паттерны загрузки (SCD1 / UPSERT)

6.1. Шаблон для справочника (пример airports_load.sql)

WITH src AS (
    SELECT
        airport_code,
        airport_name,
        city,
        country,
        coordinates,
        timezone
    FROM stg.airports
    WHERE batch_id = '{{ params.stg_batch_id }}'::text
)
UPDATE ods.airports AS o
SET airport_name = s.airport_name,
    city         = s.city,
    country      = s.country,
    coordinates  = s.coordinates,
    timezone     = s.timezone,
    _load_id     = '{{ params.stg_batch_id }}'::text,
    _load_ts     = now()
FROM src AS s
WHERE o.airport_code = s.airport_code
  AND (
      o.airport_name <> s.airport_name OR
      o.city         <> s.city OR
      o.country      <> s.country OR
      COALESCE(o.coordinates, '') <> COALESCE(s.coordinates, '') OR
      o.timezone     <> s.timezone
  );

INSERT INTO ods.airports (
    airport_code, airport_name, city, country, coordinates, timezone,
    _load_id, _load_ts
)
SELECT
    s.airport_code, s.airport_name, s.city, s.country, s.coordinates, s.timezone,
    '{{ params.stg_batch_id }}'::text, now()
FROM src s
WHERE NOT EXISTS (
    SELECT 1
    FROM ods.airports o
    WHERE o.airport_code = s.airport_code
);

ANALYZE ods.airports;

6.2. Шаблон для транзакции (пример bookings_load.sql)

WITH src AS (
    SELECT DISTINCT ON (book_ref)
        book_ref,
        book_date::TIMESTAMP WITH TIME ZONE AS book_date,
        total_amount::NUMERIC(10,2)         AS total_amount,
        src_created_at_ts                    AS event_ts
    FROM stg.bookings
    WHERE batch_id = '{{ params.stg_batch_id }}'::text
    ORDER BY book_ref, src_created_at_ts DESC NULLS LAST, load_dttm DESC
)
UPDATE ods.bookings AS o
SET book_date      = s.book_date,
    total_amount   = s.total_amount,
    event_ts       = s.event_ts,
    _load_id       = '{{ params.stg_batch_id }}'::text,
    _load_ts       = now()
FROM src AS s
WHERE o.book_ref = s.book_ref
  AND (
      o.book_date    <> s.book_date OR
      o.total_amount <> s.total_amount
  );

INSERT INTO ods.bookings (
    book_ref, book_date, total_amount, event_ts,
    _load_id, _load_ts
)
SELECT
    s.book_ref, s.book_date, s.total_amount, s.event_ts,
    '{{ params.stg_batch_id }}'::text, now()
FROM src s
WHERE NOT EXISTS (
    SELECT 1
    FROM ods.bookings o
    WHERE o.book_ref = s.book_ref
);

ANALYZE ods.bookings;

6.3. Поведение при пустом батче

• для инкрементальных таблиц (bookings, tickets, flights, segments) пустой батч допустим;
• для snapshot-справочников (airports, airplanes, routes, seats) пустой батч считаем ошибкой.

---

7) DQ-проверки ODS (минимум, но строго)

Каждый DQ-скрипт должен:

• быть привязан к stg_batch_id;
• делать RAISE EXCEPTION при нарушении;
• давать понятную подсказку в тексте ошибки.

7.1. Обязательные проверки

1. Нет дублей по бизнес-ключу в ODS.

2. Все ключи из STG текущего батча присутствуют в ODS.

3. Обязательные поля не NULL/не пустые.

4. Ссылочная целостность в ODS:

• tickets.book_ref -> bookings.book_ref
• flights.route_no -> routes.route_no
• segments.ticket_no -> tickets.ticket_no
• segments.flight_id -> flights.flight_id
• boarding_passes (ticket_no, flight_id) -> segments (ticket_no, flight_id)

7.2. Пример проверки покрытия батча

SELECT COUNT(*)
FROM (
    SELECT DISTINCT book_ref
    FROM stg.bookings
    WHERE batch_id = '{{ params.stg_batch_id }}'::text
) s
WHERE NOT EXISTS (
    SELECT 1
    FROM ods.bookings o
    WHERE o.book_ref = s.book_ref
);

Ожидаемый результат: 0.

---

8) Структура файлов

sql/ods/
├── airports_ddl.sql
├── airports_load.sql
├── airports_dq.sql
├── airplanes_ddl.sql
├── airplanes_load.sql
├── airplanes_dq.sql
├── routes_ddl.sql
├── routes_load.sql
├── routes_dq.sql
├── seats_ddl.sql
├── seats_load.sql
├── seats_dq.sql
├── bookings_ddl.sql
├── bookings_load.sql
├── bookings_dq.sql
├── tickets_ddl.sql
├── tickets_load.sql
├── tickets_dq.sql
├── flights_ddl.sql
├── flights_load.sql
├── flights_dq.sql
├── segments_ddl.sql
├── segments_load.sql
├── segments_dq.sql
├── boarding_passes_ddl.sql
├── boarding_passes_load.sql
└── boarding_passes_dq.sql

sql/ddl_gp_ods.sql

airflow/dags/
├── bookings_ods_ddl.py
└── bookings_to_gp_ods.py

docs/bookings_to_gp_ods.md
Makefile                (+ ddl-gp-ods)
tests/test_dags_smoke.py (+ smoke для 2 новых DAG)

---

9) DAG bookings_to_gp_ods: учебный граф зависимостей

Принцип: у каждой сущности строго load -> dq, и только после dq разрешаем downstream.

load_ods_bookings  -> dq_ods_bookings -> load_ods_tickets -> dq_ods_tickets

                    ├-> load_ods_airports  -> dq_ods_airports  ─┐
                    ├-> load_ods_airplanes -> dq_ods_airplanes ─┼-> load_ods_routes -> dq_ods_routes -> load_ods_flights -> dq_ods_flights
                    └->                                        └-> load_ods_seats  -> dq_ods_seats

dq_ods_flights + dq_ods_tickets -> load_ods_segments -> dq_ods_segments -> load_ods_boarding_passes -> dq_ods_boarding_passes

[dq_ods_boarding_passes, dq_ods_seats] -> finish_ods_summary

Зависимости:

• tickets после bookings;
• routes после airports и airplanes;
• seats после airplanes;
• flights после routes;
• segments после flights и tickets;
• boarding_passes после segments.

---

10) Порядок реализации

1. Подготовить DDL в sql/ods/*_ddl.sql.
2. Сделать мастер-скрипт sql/ddl_gp_ods.sql.
3. Добавить Makefile-таргет ddl-gp-ods.
4. Создать DAG bookings_ods_ddl.py.
5. Реализовать sql/ods/*_load.sql (SCD1 UPSERT).
6. Реализовать sql/ods/*_dq.sql.
7. Создать DAG bookings_to_gp_ods.py (с параметром stg_batch_id).
8. Дописать smoke-тесты DAG в tests/test_dags_smoke.py.
9. Описать запуск и проверки в docs/bookings_to_gp_ods.md.

---

11) Критерии готовности (Definition of Done)

Готово, если:

1. Оба новых DAG парсятся и проходят smoke-тесты (make test).
2. make ddl-gp-ods создаёт объекты без ошибок.
3. Для тестового stg_batch_id ODS-загрузка завершается успешно.
4. Все DQ-задачи зелёные и реально валят DAG при искусственной ошибке.
5. В ODS нет дублей по бизнес-ключам.
6. Нейминг техполей консистентен с учебной статьёй: _load_id, _load_ts, valid_from/valid_to (последние — когда перейдём к SCD2 в DDS).

---

12) Как проверять вручную

make up
make ddl-gp
# Trigger bookings_to_gp_stage
# Получить batch_id из stg.bookings (последний)
# Trigger bookings_to_gp_ods с conf: {"stg_batch_id": "<значение>"}
make gp-psql

Проверочные SQL:

-- 1) Дубликаты в ODS (пример bookings)
SELECT book_ref, COUNT(*)
FROM ods.bookings
GROUP BY 1
HAVING COUNT(*) > 1;

-- 2) Покрытие текущего STG-батча в ODS
SELECT COUNT(*)
FROM (
    SELECT DISTINCT book_ref
    FROM stg.bookings
    WHERE batch_id = '<stg_batch_id>'
) s
WHERE NOT EXISTS (
    SELECT 1
    FROM ods.bookings o
    WHERE o.book_ref = s.book_ref
);

-- 3) Ссылочная целостность tickets -> bookings
SELECT COUNT(*)
FROM ods.tickets t
WHERE NOT EXISTS (
    SELECT 1
    FROM ods.bookings b
    WHERE b.book_ref = t.book_ref
);

Ожидаемо: все три запроса возвращают 0 проблемных строк.

---

13) Что будет следующим шагом

После стабилизации ODS:

• строим DDS;
• показываем SCD2 на измерениях DDS (valid_from/valid_to, created_at/updated_at) по тому же неймингу, который уже знаком студентам из dwh-modeling.