airflow-greenplum/docs/design/bookings_stg_design.md

Дизайн STG для bookings в Greenplum

1. Цель и общий контур

• Источник: Postgres в контейнере bookings-db, база demo, таблица bookings.bookings.
• Цель: показываем путь данных от операционной БД до сырого слоя DWH в Greenplum.
• В этом документе описываем часть src (bookings-db) → STG (Greenplum). STG — входной слой; далее данные обрабатываются в ODS → DDS → DM (см. соответствующие design-документы).

Примечание: в текущей версии стенда слой stg содержит не только bookings, но и остальные таблицы потока (tickets, airports, airplanes, routes, seats, flights, segments, boarding_passes). Ниже логика разобрана на примере bookings, потому что на нём проще показать принципы инкремента и батчей.

Логика на уровне слоёв (по статье):

• src: оперативная система (bookings-db, схема bookings).
• stg: сырой слой в Greenplum, максимально близкий к источнику, без бизнес‑логики.
• далее данные обрабатываются в слоях: ods (см. bookings_ods_design.md), dds (см. bookings_dds_design.md), dm (см. bookings_dm_design.md).

2. Схема и таблицы в Greenplum

2.1. Схема

• Используем одну схему stg в Greenplum.
• В этой схеме будут:
  • внешние таблицы PXF *_ext для чтения из bookings-db;
  • внутренние таблицы STG stg.* для долговременного хранения «сырых» данных.

2.2. Внешняя таблица (PXF)

• Имя таблицы: stg.bookings_ext.
• Назначение: «окно» в исходную таблицу bookings.bookings в bookings-db через PXF (JDBC).
• Типы колонок:
  • можем использовать «родные» типы из bookings.bookings (включая даты/числа);
  • задача внешней таблицы — корректно читать данные из источника, не заниматься приведением типов.

В текущей реализации аналогично созданы внешние таблицы *_ext и для остальных сущностей (см. sql/stg/*_ddl.sql).

DDL определён в sql/stg/bookings_ddl.sql и подключается из sql/ddl_gp.sql через \i (применяется через make ddl-gp).

2.3. Внутренняя таблица STG

• Имя таблицы: stg.bookings.
• Назначение: хранить сырые данные из источника для последующей обработки (ODS/DDS/витрины).
• Принципы моделирования:
  • все бизнес‑колонки из bookings.bookings храним как TEXT (как в примерах STG из статьи);
  • не делаем UPDATE/DELETE, только INSERT новых записей;
  • бизнес‑колонки по названию совпадают с источником (чтобы проще было маппить).

Технологические колонки:

• event_ts TIMESTAMP — дата/время из источника, приведённая к TIMESTAMP:
  • используется как опорная колонка для инкрементальной загрузки;
  • заполняется из опорной даты/времени, принятой для конкретной сущности (например, для bookings — из book_date).
• _load_ts TIMESTAMP NOT NULL DEFAULT now() — когда запись была загружена в STG.
• _load_id TEXT NOT NULL — идентификатор «пачки» (например, {{ run_id }} Airflow).
• при необходимости позже можно добавить src_system TEXT, если появятся другие источники.

Колонки‑бизнес‑ключи (booking_id и т.п.) храним как TEXT. В слое DDS позже можно будет ввести суррогатные ключи и нормализовать модель под витрины.

3. Инкрементальная загрузка

3.1. Опорное поле для инкремента

• Опорная колонка: event_ts (внутреннее имя в STG).
• Источник значения:
  • для bookings используем book_date из bookings.bookings (в демо‑БД это поле естественно “шагает” по дням);
  • при чтении через stg.bookings_ext приводим к TIMESTAMP и сохраняем в stg.bookings.event_ts.

3.2. Правила определения full/delta

• При первом запуске, если таблица stg.bookings пуста:
  • считаем режим full — загружаем все строки из stg.bookings_ext.
• При последующих запусках:
  • читаем max(event_ts) из stg.bookings за все предыдущие загрузки;
  • загружаем строки, где event_ts больше этой максимальной метки (верхняя граница по времени не задаётся).

Таким образом, вся логика инкремента «замкнута» на один техно‑столбец event_ts, который студент потом сможет использовать и на следующих слоях (например, в CDC‑логике).

4. DAG’и Airflow (логика на уровне задач)

4.1. DAG для DDL

• dag_id: bookings_stg_ddl (реализован в airflow/dags/bookings_stg_ddl.py).
• Назначение: один раз (или при изменении схемы) создать необходимые объекты в Greenplum:
  • схему stg (если её ещё нет);
  • внешние таблицы *_ext и внутренние таблицы слоя stg для всех сущностей потока (см. sql/stg/*_ddl.sql).
• Этот DAG не загружает данные, только подготавливает структуру.
• Вся DDL‑логика (CREATE/ALTER/DROP) сосредоточена здесь; рабочие DAG’и занимаются только DML (INSERT/SELECT).

4.2. DAG для пошаговой загрузки

• dag_id: bookings_to_gp_stage.
• Основные параметры:
  • _load_id (в текущей реализации {{ run_id }}) — метка батча, которая попадает в stg.bookings._load_id;
  • подключения:
    • bookings_db_conn_id — Airflow connection к bookings-db (в коде DAG — BOOKINGS_CONN_ID = "bookings_db");
    • greenplum_conn_id — Airflow connection к Greenplum (GREENPLUM_CONN_ID = "greenplum_conn").

Последовательность задач (упрощённая, но отражающая те же шаги):

1. generate_bookings_day
   • PostgresOperator к bookings-db;
   • выполняет скрипт /sql/src/bookings_generate_day_if_missing.sql;
   • скрипт смотрит на max(book_date) и:
     • если база пуста — берёт стартовую дату из конфигурации (bookings.start_date) и генерирует bookings.init_days суток;
     • если данные уже есть — добавляет один следующий учебный день после max(book_date) (логика как в bookings/generate_next_day.sql).
2. load_bookings_to_stg
   • PostgresOperator к Greenplum;
   • выполняет скрипт /sql/stg/bookings_load.sql;
   • внутри SQL считается max(event_ts) по «старым» батчам и по нему строится окно инкремента:
     • первая загрузка (full) — берём все строки из stg.bookings_ext;
     • последующие загрузки — берём только записи, где book_date больше предыдущего максимума (верхняя граница по дате не задаётся явно);
   • при вставке заполняются тех.колонки event_ts, _load_ts, _load_id.
3. check_row_counts
   • PostgresOperator к Greenplum;
   • выполняет скрипт /sql/stg/bookings_dq.sql;
   • скрипт заново считает окно инкремента по тем же правилам, что и загрузка, и сравнивает:
     • количество строк в stg.bookings_ext с book_date позже «старого» максимума,
     • количество строк в stg.bookings для текущего _load_id;
   • при расхождении выполняет RAISE EXCEPTION с понятным текстом ошибки.
4. Далее — загрузка и DQ для остальных таблиц потока (tickets, справочники, транзакции).
5. finish_summary
   • PythonOperator, который логирует итог выполнения DAG и напоминает, где смотреть детальные логи.

Таким образом, вся бизнес‑логика инкремента и проверок живёт в SQL‑скриптах, а DAG отвечает за оркестрацию и подключение к нужным БД. Для менти это хороший пример разделения ответственности между SQL и Python.

5. Связь с остальными документами

• Детали настройки PXF и часовых поясов — в ветке solution (docs/reference/).
• sql/stg/bookings_ddl.sql — DDL для схемы stg и таблиц stg.bookings_ext / stg.bookings (подключается из sql/ddl_gp.sql и применяется через make ddl-gp).

Дальнейшая обработка данных описана в design-документах ODS/DDS/DM (см. раздел 5).

6. Требования к читаемости и комментариям

• DAG’и bookings_stg_ddl и bookings_to_gp_stage — это учебный материал для менти.
• В коде DAG’ов должны быть:
  • понятные docstring на русском у всех функций и DAG;
  • короткие комментарии рядом с нетривиальной логикой (особенно вокруг инкремента и идемпотентности);
  • говорящие task_id и названия функций, отражающие их роль в процессе.
• Цель: чтобы по одному только коду DAG студент мог восстановить архитектуру процесса и сопоставить её с теорией из статьи про моделирование DWH.