diff --git a/README.md b/README.md index 07f59ac..5b79da6 100644 --- a/README.md +++ b/README.md @@ -84,7 +84,7 @@ make gp-psql -- внутри psql: SELECT COUNT(*) FROM stg.bookings; SELECT COUNT(*) FROM stg.tickets; -SELECT * FROM stg.bookings ORDER BY src_created_at_ts DESC LIMIT 10; +SELECT * FROM stg.bookings ORDER BY event_ts DESC LIMIT 10; SELECT COUNT(*) FROM ods.bookings; SELECT COUNT(*) FROM ods.tickets; SELECT COUNT(*) FROM dds.dim_routes; diff --git a/TESTING.md b/TESTING.md index 67747f4..a780b26 100644 --- a/TESTING.md +++ b/TESTING.md @@ -39,7 +39,7 @@ - важно: DAG `bookings_stg_ddl` **не** создаёт базу `demo` в `bookings-db`; если вы делали `docker compose down -v` / `make clean`, `make bookings-init` обязателен (быстрое восстановление из seed-дампа); - включить DAG `bookings_to_gp_stage` и запустить `Trigger DAG`; - убедиться, что все задачи завершились со статусом Success (включая загрузки справочников/транзакций и DQ); - - при желании проверить данные: в `bookings-db` появился новый день, а в Greenplum в `stg.bookings` — строки с актуальным `batch_id` (см. пример запросов в разделе 5). + - при желании проверить данные: в `bookings-db` появился новый день, а в Greenplum в `stg.bookings` — строки с актуальным `_load_id` (см. пример запросов в разделе 5). - (опционально, для менторов/разработчиков) Smoke-тест DAG через Airflow CLI без UI: - `docker compose -f docker-compose.yml exec airflow-webserver airflow dags test bookings_to_gp_stage 2024-01-01` — прогоняет `bookings_to_gp_stage` целиком в «off-line» режиме; diff --git a/airflow/dags/bookings_to_gp_ods.py b/airflow/dags/bookings_to_gp_ods.py index 1acd63a..a38287d 100644 --- a/airflow/dags/bookings_to_gp_ods.py +++ b/airflow/dags/bookings_to_gp_ods.py @@ -39,7 +39,7 @@ def _resolve_stg_batch_id(**context) -> str: Для инкрементальных транзакционных таблиц (bookings, tickets, flights, segments, boarding_passes) этот батч НЕ используется. Вместо этого они грузят все новые - записи по HWM: WHERE load_dttm > (SELECT MAX(_load_ts) FROM ods.table). + записи по HWM: WHERE _load_ts > (SELECT MAX(_load_ts) FROM ods.table). Это сделано для того, чтобы не потерять инкременты, если STG-DAG запускался несколько раз до запуска ODS-DAG'а. @@ -57,50 +57,50 @@ def _resolve_stg_batch_id(**context) -> str: result = hook.get_first( """ WITH candidate_batches AS ( - SELECT batch_id + SELECT _load_id FROM stg.airports - WHERE batch_id IS NOT NULL AND batch_id <> '' - GROUP BY batch_id + WHERE _load_id IS NOT NULL AND _load_id <> '' + GROUP BY _load_id INTERSECT - SELECT batch_id + SELECT _load_id FROM stg.airplanes - WHERE batch_id IS NOT NULL AND batch_id <> '' - GROUP BY batch_id + WHERE _load_id IS NOT NULL AND _load_id <> '' + GROUP BY _load_id INTERSECT - SELECT batch_id + SELECT _load_id FROM stg.routes - WHERE batch_id IS NOT NULL AND batch_id <> '' - GROUP BY batch_id + WHERE _load_id IS NOT NULL AND _load_id <> '' + GROUP BY _load_id INTERSECT - SELECT batch_id + SELECT _load_id FROM stg.seats - WHERE batch_id IS NOT NULL AND batch_id <> '' - GROUP BY batch_id + WHERE _load_id IS NOT NULL AND _load_id <> '' + GROUP BY _load_id ), batch_ready AS ( SELECT - c.batch_id, + c._load_id, GREATEST( COALESCE( - (SELECT MAX(load_dttm) FROM stg.airports a WHERE a.batch_id = c.batch_id), + (SELECT MAX(_load_ts) FROM stg.airports a WHERE a._load_id = c._load_id), TIMESTAMP '1900-01-01 00:00:00' ), COALESCE( - (SELECT MAX(load_dttm) FROM stg.airplanes a WHERE a.batch_id = c.batch_id), + (SELECT MAX(_load_ts) FROM stg.airplanes a WHERE a._load_id = c._load_id), TIMESTAMP '1900-01-01 00:00:00' ), COALESCE( - (SELECT MAX(load_dttm) FROM stg.routes r WHERE r.batch_id = c.batch_id), + (SELECT MAX(_load_ts) FROM stg.routes r WHERE r._load_id = c._load_id), TIMESTAMP '1900-01-01 00:00:00' ), COALESCE( - (SELECT MAX(load_dttm) FROM stg.seats s WHERE s.batch_id = c.batch_id), + (SELECT MAX(_load_ts) FROM stg.seats s WHERE s._load_id = c._load_id), TIMESTAMP '1900-01-01 00:00:00' ) ) AS ready_dttm FROM candidate_batches c ) - SELECT batch_id + SELECT _load_id FROM batch_ready ORDER BY ready_dttm DESC LIMIT 1 diff --git a/airflow/dags/bookings_to_gp_stage.py b/airflow/dags/bookings_to_gp_stage.py index c6dc0bf..ad7b8bb 100644 --- a/airflow/dags/bookings_to_gp_stage.py +++ b/airflow/dags/bookings_to_gp_stage.py @@ -12,7 +12,7 @@ from __future__ import annotations Каждый запуск DAG работает как «шаг по времени вперёд»: - генератор в демо-БД bookings добавляет следующий учебный день после max(book_date); - загрузка в Greenplum берёт все строки, появившиеся после предыдущих батчей; -- `run_id` используется как метка запуска (в `batch_id`, в логах и DQ). +- `run_id` используется как метка запуска (в `_load_id`, в логах и DQ). Важно: для инкрементальных таблиц «пустое окно инкремента» допустимо (это не ошибка). diff --git a/docs/bookings_to_gp_ods.md b/docs/bookings_to_gp_ods.md index 53c7259..ecb47d3 100644 --- a/docs/bookings_to_gp_ods.md +++ b/docs/bookings_to_gp_ods.md @@ -7,7 +7,7 @@ - Определяет `stg_batch_id`: - берёт из `dag_run.conf["stg_batch_id"]`, если передан; - - иначе берёт последний **согласованный** `batch_id`, который есть во всех snapshot-таблицах STG + - иначе берёт последний **согласованный** `_load_id`, который есть во всех snapshot-таблицах STG (`airports`, `airplanes`, `routes`, `seats`). - Загружает 9 таблиц ODS (`airports`, `airplanes`, `routes`, `seats`, `bookings`, `tickets`, `flights`, `segments`, `boarding_passes`). diff --git a/docs/bookings_to_gp_stage.md b/docs/bookings_to_gp_stage.md index 335a604..ea6ba90 100644 --- a/docs/bookings_to_gp_stage.md +++ b/docs/bookings_to_gp_stage.md @@ -63,15 +63,15 @@ make bookings-init - выполняет `sql/stg/bookings_load.sql` в Greenplum; - берёт строки из `stg.bookings_ext`, которые попадают в новое окно инкремента; - вставляет их в `stg.bookings`, добавляя тех.колонки: - - `src_created_at_ts` (опорная метка времени для инкремента), - - `load_dttm`, - - `batch_id={{ run_id }}`. + - `event_ts` (опорная метка времени для инкремента), + - `_load_ts`, + - `_load_id={{ run_id }}`. 3) `check_row_counts` - выполняет `sql/stg/bookings_dq.sql` в Greenplum; - считает количество строк в источнике за то же окно инкремента и сравнивает с количеством строк, - вставленных в `stg.bookings` для текущего `batch_id`; + вставленных в `stg.bookings` для текущего `_load_id`; - при расхождении делает `RAISE EXCEPTION` с понятным текстом. 4) `load_tickets_to_stg` @@ -79,7 +79,7 @@ make bookings-init - выполняет `sql/stg/tickets_load.sql` в Greenplum; - так как в `bookings.tickets` нет явной временной колонки, окно инкремента берётся по `book_date` из связанной внешней таблицы `stg.bookings_ext` (JOIN по `book_ref`); -- вставляет строки в `stg.tickets`, добавляя `src_created_at_ts`, `load_dttm` и `batch_id={{ run_id }}`. +- вставляет строки в `stg.tickets`, добавляя `event_ts`, `_load_ts` и `_load_id={{ run_id }}`. 5) `check_tickets_dq` @@ -135,11 +135,11 @@ make gp-psql SELECT COUNT(*) FROM stg.bookings; SELECT - src_created_at_ts, - load_dttm, - batch_id + event_ts, + _load_ts, + _load_id FROM stg.bookings -ORDER BY src_created_at_ts DESC +ORDER BY event_ts DESC LIMIT 10; ``` diff --git a/docs/internal/PRD.md b/docs/internal/PRD.md index 7db5630..b61222e 100644 --- a/docs/internal/PRD.md +++ b/docs/internal/PRD.md @@ -53,7 +53,7 @@ end-to-end ETL-пайплайн: от базы-источника до анал (STG → ODS → DDS → DM) на реальном стеке Airflow + Greenplum. 2. **Читать ТЗ от аналитика** (маппинги, описания таблиц) и превращать его в работающий SQL + DAG. -3. **Писать идемпотентные загрузки** с инкрементальностью (HWM, batch_id, +3. **Писать идемпотентные загрузки** с инкрементальностью (HWM, _load_id, delete+insert), понимая, почему в Greenplum не используется MERGE. 4. **Реализовывать SCD1/SCD2** и объяснять, когда что применяется. 5. **Настраивать и проверять Data Quality** — понимает, зачем DQ-проверки @@ -210,7 +210,7 @@ solution (полное решение) - [ ] SQL идемпотентен (повторный запуск не ломает данные) - [ ] Distribution keys выбраны осмысленно - [ ] Студент может объяснить: почему delete+insert, а не MERGE; - разницу SCD1/SCD2; что такое HWM; как работает batch_id + разницу SCD1/SCD2; что такое HWM; как работает _load_id - [ ] Код оформлен для портфолио (чистый Git-history, README) --- diff --git a/docs/internal/architecture_review.md b/docs/internal/architecture_review.md index 0a5276c..124c8e3 100644 --- a/docs/internal/architecture_review.md +++ b/docs/internal/architecture_review.md @@ -45,7 +45,7 @@ LEFT JOIN dds.dim_routes AS rte GP-специфичная best practice, которую забывают даже опытные команды. ### 9. Идемпотентные STG-загрузки -`NOT EXISTS (... WHERE batch_id = '{{ run_id }}')` — простой, корректный, понятный паттерн для retry-safe загрузок. +`NOT EXISTS (... WHERE _load_id = '{{ run_id }}')` — простой, корректный, понятный паттерн для retry-safe загрузок. --- @@ -56,7 +56,7 @@ GP-специфичная best practice, которую забывают даж - [x] **ODS batch resolver теряет данные при двух STG-запусках подряд** - Сценарий: STG run_1 загружает день N, STG run_2 загружает день N+1, затем ODS запускается - `_resolve_stg_batch_id()` выбирает только последний согласованный batch (`run_2`) - - Все ODS load-скрипты фильтруют `WHERE batch_id = 'run_2'` → данные `run_1` навсегда пропущены + - Все ODS load-скрипты фильтруют `WHERE _load_id = 'run_2'` → данные `run_1` навсегда пропущены - **Справочники** (airports, airplanes, routes, seats): проблемы нет — full snapshot, `run_2` содержит всё - **Транзакционные таблицы** (bookings, tickets, flights, segments, boarding_passes): **потеря данных** — инкрементальные записи `run_1` никогда не попадут в ODS - Корень проблемы: batch resolver проектировался для согласованности справочников (INTERSECT), но тот же single-batch фильтр применяется к транзакционным таблицам, где нужны **все необработанные** batch-и @@ -117,12 +117,11 @@ GP-специфичная best practice, которую забывают даж - **Решение**: вынесено в CREATE TEMP TABLE tmp_routes_src ON COMMIT DROP ✅ ВЫПОЛНЕНО - Файл: `sql/dds/dim_routes_load.sql` -- [ ] **Несогласованность нейминга STG vs ODS+** ✅ РЕШЕНИЕ ПРИНЯТО - - STG: `batch_id`, `load_dttm`, `src_created_at_ts` → переименовать в канон `_load_id`, `_load_ts`, `event_ts` +- [x] **Несогласованность нейминга STG vs ODS+** ✅ ВЫПОЛНЕНО + - STG: `batch_id`, `load_dttm`, `src_created_at_ts` → переименованы в канон `_load_id`, `_load_ts`, `event_ts` - Единый словарь во всех слоях снижает когнитивную нагрузку - - Добавить заметку в `naming_conventions.md` (секция «legacy-нейминг в реальных проектах») - - Удалить секцию 6 «Переходный маппинг» как неактуальную - - Файлы: ~27 STG SQL + ODS load-скрипты + `naming_conventions.md` + тесты + - Секция 6 «Переходный маппинг» удалена из `naming_conventions.md` как неактуальная + - Файлы: 27 STG SQL + ODS load-скрипты + `naming_conventions.md` + тесты - [x] **Дублирование CTE в ODS load-скриптах** - `WITH src AS (...)` копируется 2-3 раза в каждом из 9 ODS load-файлов @@ -149,7 +148,7 @@ GP-специфичная best practice, которую забывают даж - [ ] **Отсутствующие паттерны** (комментарии/заметки): - Partitioning (когда и зачем, почему не здесь) - SCD Type 3/6 (хотя бы упомянуть существование) - - Data lineage (`_load_id` в DM ≠ `batch_id` в STG — нет сквозного трассирования) + - Data lineage (сквозное трассирование `_load_id` через STG → ODS → DDS → DM) --- @@ -296,7 +295,7 @@ DROP TABLE tmp_fact_20170102; | ~~P1~~ | ~~Добавить 7 точечных комментариев~~ | ~~30-40 мин~~ | ~~done~~ | | **P2** | Явный storage type + AO где нет UPDATE (ADR-3) | 2-3 часа | все `*_ddl.sql` в ods/dds/dm + 4 ODS snapshot load | | **P2** | Рефакторинг hashdiff → TEMP TABLE | 1 час | `sql/dds/dim_routes_load.sql` | -| **P2** | Переименовать STG поля в канон + заметка | 1-2 часа | 27 STG SQL + ODS load + naming_conventions.md | +| ~~P2~~ | ~~Переименовать STG поля в канон + заметка~~ | ~~1-2 часа~~ | ~~done~~ | | **P2** | TEMP TABLE для сложных ODS load-ов | 1 час | 3-4 ODS load файла | | **P2** | Реализовать `dm.route_performance` | 2-3 часа | 3 SQL + DAG + тесты | | **P3** | Маршрут изучения DDS + distribution strategy doc | 1 час | 2 новых md-файла | diff --git a/docs/internal/bookings_ods_design.md b/docs/internal/bookings_ods_design.md index d6960c5..4c1870b 100644 --- a/docs/internal/bookings_ods_design.md +++ b/docs/internal/bookings_ods_design.md @@ -5,7 +5,7 @@ STG-слой уже реализован как учебный эталон: - данные из `bookings-db` читаются через PXF; - в STG бизнес-колонки хранятся как `TEXT`; -- загрузка и DQ работают батчами (`batch_id = {{ run_id }}`). +- загрузка и DQ работают батчами (`_load_id = {{ run_id }}`). Этот документ фиксирует **простую и каноничную** реализацию ODS для менти. @@ -36,7 +36,7 @@ ODS в учебном проекте — это: ### 1.3. Где хранится история изменений -- История «как приходили данные» уже сохраняется в STG (append + `batch_id`). +- История «как приходили данные» уже сохраняется в STG (append + `_load_id`). - Историзацию измерений (SCD2) показываем позже в DDS (как в учебной статье `dwh-modeling`). Итог: **ODS = текущий слой (current state), простой и понятный**. @@ -55,9 +55,9 @@ ODS в учебном проекте — это: ### 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` (для транзакционных таблиц) +- `stg._load_id` -> `ods._load_id` +- `stg._load_ts` не переносим 1:1; в ODS пишем собственный `ods._load_ts = now()` +- `stg.event_ts` -> `ods.event_ts` (для транзакционных таблиц) ### 2.2. Почему так @@ -239,8 +239,8 @@ DISTRIBUTED BY (ticket_no) | `flights` | `flight_id` | `flight_id` | `INTEGER` | Только cast TEXT → INT | | `segments` | `flight_id` | `flight_id` | `INTEGER` | Только cast TEXT → INT | | `boarding_passes` | `flight_id` | `flight_id` | `INTEGER` | Только cast TEXT → INT | -| все транзакционные | `src_created_at_ts` | `event_ts` | `TIMESTAMP` | Маппинг legacy → канон | -| все | `batch_id` | `_load_id` | `TEXT` | Маппинг legacy → канон | +| все транзакционные | `event_ts` | `event_ts` | `TIMESTAMP` | Прямой перенос (канон) | +| все | `_load_id` | `_load_id` | `TEXT` | Прямой перенос (канон) | Пример каста с переименованием в SQL (в CTE): ```sql @@ -282,32 +282,32 @@ def _resolve_stg_batch_id(**context): stg_batch_id = conf.get("stg_batch_id") if not stg_batch_id: - # Берём batch_id, который присутствует во всех snapshot-таблицах STG: + # Берём _load_id, который присутствует во всех snapshot-таблицах STG: # airports, airplanes, routes, seats. Это защищает от частично успешных запусков. hook = PostgresHook(postgres_conn_id=GREENPLUM_CONN_ID) result = hook.get_first( ''' WITH candidate_batches AS ( - SELECT batch_id FROM stg.airports WHERE batch_id IS NOT NULL GROUP BY batch_id + SELECT _load_id FROM stg.airports WHERE _load_id IS NOT NULL GROUP BY _load_id INTERSECT - SELECT batch_id FROM stg.airplanes WHERE batch_id IS NOT NULL GROUP BY batch_id + SELECT _load_id FROM stg.airplanes WHERE _load_id IS NOT NULL GROUP BY _load_id INTERSECT - SELECT batch_id FROM stg.routes WHERE batch_id IS NOT NULL GROUP BY batch_id + SELECT _load_id FROM stg.routes WHERE _load_id IS NOT NULL GROUP BY _load_id INTERSECT - SELECT batch_id FROM stg.seats WHERE batch_id IS NOT NULL GROUP BY batch_id + SELECT _load_id FROM stg.seats WHERE _load_id IS NOT NULL GROUP BY _load_id ), batch_ready AS ( SELECT - c.batch_id, + c._load_id, GREATEST( - (SELECT MAX(load_dttm) FROM stg.airports a WHERE a.batch_id = c.batch_id), - (SELECT MAX(load_dttm) FROM stg.airplanes a WHERE a.batch_id = c.batch_id), - (SELECT MAX(load_dttm) FROM stg.routes r WHERE r.batch_id = c.batch_id), - (SELECT MAX(load_dttm) FROM stg.seats s WHERE s.batch_id = c.batch_id) + (SELECT MAX(_load_ts) FROM stg.airports a WHERE a._load_id = c._load_id), + (SELECT MAX(_load_ts) FROM stg.airplanes a WHERE a._load_id = c._load_id), + (SELECT MAX(_load_ts) FROM stg.routes r WHERE r._load_id = c._load_id), + (SELECT MAX(_load_ts) FROM stg.seats s WHERE s._load_id = c._load_id) ) AS ready_dttm FROM candidate_batches c ) - SELECT batch_id + SELECT _load_id FROM batch_ready ORDER BY ready_dttm DESC LIMIT 1 @@ -334,7 +334,7 @@ resolve_batch = PythonOperator( Во всех `sql/ods/*_load.sql` и `sql/ods/*_dq.sql` значение `stg_batch_id` подставляется через Jinja-шаблон: ```sql -WHERE batch_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text +WHERE _load_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text ``` Для читаемости в DAG можно вынести шаблон в константу: @@ -370,7 +370,7 @@ WITH src AS ( coordinates, timezone FROM stg.airports - WHERE batch_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text + WHERE _load_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text ) UPDATE ods.airports AS o SET airport_name = s.airport_name, @@ -403,7 +403,7 @@ WITH src AS ( coordinates, timezone FROM stg.airports - WHERE batch_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text + WHERE _load_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text ) INSERT INTO ods.airports ( airport_code, airport_name, city, country, coordinates, timezone, @@ -423,7 +423,7 @@ WHERE NOT EXISTS ( WITH src_keys AS ( SELECT DISTINCT airport_code FROM stg.airports - WHERE batch_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text + WHERE _load_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text ) DELETE FROM ods.airports o WHERE NOT EXISTS ( @@ -447,13 +447,13 @@ WITH src AS ( 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, + event_ts, ROW_NUMBER() OVER ( PARTITION BY book_ref - ORDER BY src_created_at_ts DESC NULLS LAST, load_dttm DESC + ORDER BY event_ts DESC NULLS LAST, _load_ts DESC ) AS rn FROM stg.bookings - WHERE batch_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text + WHERE _load_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text ) UPDATE ods.bookings AS o SET book_date = s.book_date, @@ -475,13 +475,13 @@ WITH src AS ( 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, + event_ts, ROW_NUMBER() OVER ( PARTITION BY book_ref - ORDER BY src_created_at_ts DESC NULLS LAST, load_dttm DESC + ORDER BY event_ts DESC NULLS LAST, _load_ts DESC ) AS rn FROM stg.bookings - WHERE batch_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text + WHERE _load_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text ) INSERT INTO ods.bookings ( book_ref, book_date, total_amount, event_ts, @@ -547,7 +547,7 @@ SELECT COUNT(*) FROM ( SELECT DISTINCT book_ref FROM stg.bookings - WHERE batch_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text + WHERE _load_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text ) s WHERE NOT EXISTS ( SELECT 1 @@ -694,7 +694,7 @@ SELECT COUNT(*) FROM ( SELECT DISTINCT book_ref FROM stg.bookings - WHERE batch_id = '' + WHERE _load_id = '' ) s WHERE NOT EXISTS ( SELECT 1 diff --git a/docs/internal/bookings_stg_code_review.md b/docs/internal/bookings_stg_code_review.md index 8a909b2..6fc2819 100644 --- a/docs/internal/bookings_stg_code_review.md +++ b/docs/internal/bookings_stg_code_review.md @@ -44,16 +44,16 @@ ### 2.2. DQ-проверки ссылочной целостности: “текущий батч” vs “вся история” (статус: исправлено) -Часть DQ-скриптов проверяет наличие “родительских” записей в таблице **без фильтра `batch_id`**. +Часть DQ-скриптов проверяет наличие “родительских” записей в таблице **без фильтра `_load_id`**. При append-only истории это может скрыть проблемы текущей загрузки: родитель был загружен в прошлом батче → проверка пройдёт, даже если текущий батч родителя не загрузил. Что сделано: -- `routes_dq.sql`: проверка airports/airplanes стала батч-строгой (`batch_id = текущий батч`). -- `seats_dq.sql`: проверка airplanes стала батч-строгой (`batch_id = текущий батч`). -- `flights_dq.sql`: проверка routes стала батч-строгой (`batch_id = текущий батч`). +- `routes_dq.sql`: проверка airports/airplanes стала батч-строгой (`_load_id = текущий батч`). +- `seats_dq.sql`: проверка airplanes стала батч-строгой (`_load_id = текущий батч`). +- `flights_dq.sql`: проверка routes стала батч-строгой (`_load_id = текущий батч`). -Примечание (почему не везде `batch_id = текущий батч`): +Примечание (почему не везде `_load_id = текущий батч`): - Если дочерняя таблица грузится инкрементом, то ссылки могут указывать на “исторические” записи, загруженные в предыдущих батчах → для таких связей корректнее проверять “существует в STG вообще”. - Для `boarding_passes` (full snapshot) ссылки на `tickets/segments` также проверяются по STG-истории, @@ -94,11 +94,11 @@ Типовой паттерн: ```sql WHERE NOT EXISTS ( - SELECT 1 FROM stg.table WHERE batch_id = '{{ run_id }}' AND key = ext.key + SELECT 1 FROM stg.table WHERE _load_id = '{{ run_id }}' AND key = ext.key ); ``` -Это в первую очередь защита от повторного запуска того же таска в рамках одного `batch_id` (retry), +Это в первую очередь защита от повторного запуска того же таска в рамках одного `_load_id` (retry), а не “лечение” дублей в источнике. Что сделано: @@ -126,7 +126,7 @@ WHERE NOT EXISTS ( ```sql LEFT JOIN stg.airports AS a ON r.departure_airport = a.airport_code - AND a.batch_id = v_batch_id + AND a._load_id = v_batch_id ``` ### 4.2. Smoke-тест реального графа (минимальный полезный уровень) diff --git a/docs/internal/bookings_stg_design.md b/docs/internal/bookings_stg_design.md index 7e1750d..e0e6da5 100644 --- a/docs/internal/bookings_stg_design.md +++ b/docs/internal/bookings_stg_design.md @@ -48,11 +48,11 @@ DDL определён в `sql/stg/bookings_ddl.sql` и подключается Технологические колонки: -- `src_created_at_ts TIMESTAMP` — дата/время из источника, приведённая к TIMESTAMP: +- `event_ts TIMESTAMP` — дата/время из источника, приведённая к TIMESTAMP: - используется как опорная колонка для инкрементальной загрузки; - заполняется из опорной даты/времени, принятой для конкретной сущности (например, для `bookings` — из `book_date`). -- `load_dttm TIMESTAMP NOT NULL DEFAULT now()` — когда запись была загружена в STG. -- `batch_id TEXT NOT NULL` — идентификатор «пачки» (например, `{{ ds_nodash }}` или `run_id` Airflow). +- `_load_ts TIMESTAMP NOT NULL DEFAULT now()` — когда запись была загружена в STG. +- `_load_id TEXT NOT NULL` — идентификатор «пачки» (например, `{{ run_id }}` Airflow). - при необходимости позже можно добавить `src_system TEXT`, если появятся другие источники. Колонки‑бизнес‑ключи (`booking_id` и т.п.) храним как `TEXT`. В слое DDS позже можно будет ввести суррогатные ключи и нормализовать модель под витрины. @@ -61,20 +61,20 @@ DDL определён в `sql/stg/bookings_ddl.sql` и подключается ### 3.1. Опорное поле для инкремента -- Опорная колонка: `src_created_at_ts` (внутреннее имя в STG). +- Опорная колонка: `event_ts` (внутреннее имя в STG). - Источник значения: - для `bookings` используем `book_date` из `bookings.bookings` (в демо‑БД это поле естественно “шагает” по дням); - - при чтении через `stg.bookings_ext` приводим к `TIMESTAMP` и сохраняем в `stg.bookings.src_created_at_ts`. + - при чтении через `stg.bookings_ext` приводим к `TIMESTAMP` и сохраняем в `stg.bookings.event_ts`. ### 3.2. Правила определения full/delta - При первом запуске, если таблица `stg.bookings` пуста: - считаем режим `full` — загружаем все строки из `stg.bookings_ext`. - При последующих запусках: - - читаем `max(src_created_at_ts)` из `stg.bookings` за все предыдущие загрузки; - - загружаем строки, где `src_created_at_ts` больше этой максимальной метки (верхняя граница по времени не задаётся). + - читаем `max(event_ts)` из `stg.bookings` за все предыдущие загрузки; + - загружаем строки, где `event_ts` больше этой максимальной метки (верхняя граница по времени не задаётся). -Таким образом, вся логика инкремента «замкнута» на один техно‑столбец `src_created_at_ts`, который студент потом сможет использовать и на следующих слоях (например, в CDC‑логике). +Таким образом, вся логика инкремента «замкнута» на один техно‑столбец `event_ts`, который студент потом сможет использовать и на следующих слоях (например, в CDC‑логике). ## 4. DAG’и Airflow (логика на уровне задач) @@ -92,7 +92,7 @@ DDL определён в `sql/stg/bookings_ddl.sql` и подключается - `dag_id`: `bookings_to_gp_stage`. - Основные параметры: - - `batch_id` (в текущей реализации `{{ run_id }}`) — метка батча, которая попадает в `stg.bookings.batch_id`; + - `_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"`). @@ -108,16 +108,16 @@ DDL определён в `sql/stg/bookings_ddl.sql` и подключается 2. `load_bookings_to_stg` - PostgresOperator к Greenplum; - выполняет скрипт `/sql/stg/bookings_load.sql`; - - внутри SQL считается `max(src_created_at_ts)` по «старым» батчам и по нему строится окно инкремента: + - внутри SQL считается `max(event_ts)` по «старым» батчам и по нему строится окно инкремента: - первая загрузка (full) — берём все строки из `stg.bookings_ext`; - последующие загрузки — берём только записи, где `book_date` больше предыдущего максимума (верхняя граница по дате не задаётся явно); - - при вставке заполняются тех.колонки `src_created_at_ts`, `load_dttm`, `batch_id`. + - при вставке заполняются тех.колонки `event_ts`, `_load_ts`, `_load_id`. 3. `check_row_counts` - PostgresOperator к Greenplum; - выполняет скрипт `/sql/stg/bookings_dq.sql`; - скрипт заново считает окно инкремента по тем же правилам, что и загрузка, и сравнивает: - количество строк в `stg.bookings_ext` с `book_date` позже «старого» максимума, - - количество строк в `stg.bookings` для текущего `batch_id`; + - количество строк в `stg.bookings` для текущего `_load_id`; - при расхождении выполняет `RAISE EXCEPTION` с понятным текстом ошибки. 4. Далее — загрузка и DQ для остальных таблиц потока (tickets, справочники, транзакции). 5. `finish_summary` diff --git a/docs/internal/db_schema.md b/docs/internal/db_schema.md index d53c343..e656efa 100644 --- a/docs/internal/db_schema.md +++ b/docs/internal/db_schema.md @@ -40,11 +40,11 @@ - **Назначение**: Сырой слой, максимально близкий к источнику, без бизнес-логики - **Хранение**: AO-Row (Append-Only Row-oriented) для эффективной загрузки больших объёмов - **Типы данных**: Бизнес-колонки как `TEXT`, тех.колонки как `TIMESTAMP` -- **Инкрементальная загрузка**: Опорное поле `src_created_at_ts` (из `book_date` для tickets) +- **Инкрементальная загрузка**: Опорное поле `event_ts` (из `book_date` для tickets) - **Технологические колонки**: - - `src_created_at_ts TIMESTAMP` — дата/время из источника для инкремента - - `load_dttm TIMESTAMP NOT NULL DEFAULT now()` — когда запись была загружена - - `batch_id TEXT` — идентификатор пачки (рекомендуем `NOT NULL`, например `{{ ds_nodash }}` или `{{ run_id }}`) + - `event_ts TIMESTAMP` — дата/время из источника для инкремента + - `_load_ts TIMESTAMP NOT NULL DEFAULT now()` — когда запись была загружена + - `_load_id TEXT` — идентификатор пачки/батча (например `{{ run_id }}`) - **DQ-проверки (после загрузки STG)**: отдельные SQL-скрипты, которые валидируют данные (counts, дубли, NULL, orphan records) и при ошибке делают `RAISE EXCEPTION`; примеры: `sql/stg/bookings_dq.sql`, `sql/stg/tickets_dq.sql` #### ODS (Operational Data Store) @@ -105,7 +105,7 @@ - `book_ref TEXT` - номер бронирования - `book_date TEXT` - дата бронирования - `total_amount TEXT` - общая сумма -- **Технические колонки:** `src_created_at_ts` (=book_date), `load_dttm`, `batch_id` +- **Технические колонки:** `event_ts` (=book_date), `_load_ts`, `_load_id` - **Стратегия загрузки:** Инкремент по `book_date` - **DQ проверки:** count (окно инкремента, пустое окно допустимо), дубликаты book_ref, NULL обязательных полей @@ -119,7 +119,7 @@ - `passenger_id TEXT` - идентификатор пассажира - `passenger_name TEXT` - имя пассажира - `outbound TEXT` - направление (в источнике boolean) -- **Технические колонки:** `src_created_at_ts` (из book_date через bookings), `load_dttm`, `batch_id` +- **Технические колонки:** `event_ts` (из book_date через bookings), `_load_ts`, `_load_id` - **Стратегия загрузки:** Инкремент по `book_date` (через bookings) - **DQ проверки:** count (окно инкремента, пустое окно допустимо), дубликаты ticket_no, NULL обязательных полей, пустой passenger_name, ссылочная целостность (bookings) @@ -133,7 +133,7 @@ - `country TEXT` - страна (из JSONB) - `coordinates TEXT` - координаты - `timezone TEXT` - часовой пояс -- **Технические колонки:** `src_created_at_ts`, `load_dttm`, `batch_id` +- **Технические колонки:** `event_ts` (=now()), `_load_ts`, `_load_id` - **Стратегия загрузки:** Full load (все строки при каждом запуске) - **DQ проверки:** count, дубликаты airport_code, NULL обязательных полей @@ -145,7 +145,7 @@ - `model TEXT` - модель (из JSONB) - `range TEXT` - дальность полёта - `speed TEXT` - скорость -- **Технические колонки:** `src_created_at_ts`, `load_dttm`, `batch_id` +- **Технические колонки:** `event_ts` (=now()), `_load_ts`, `_load_id` - **Стратегия загрузки:** Full load - **DQ проверки:** count, дубликаты airplane_code, NULL обязательных полей @@ -162,9 +162,9 @@ - `days_of_week TEXT` - дни недели (из int[]) - `scheduled_time TEXT` - плановое время - `duration TEXT` - длительность -- **Технические колонки:** `src_created_at_ts`, `load_dttm`, `batch_id` +- **Технические колонки:** `event_ts` (=now()), `_load_ts`, `_load_id` - **Стратегия загрузки:** Full load -- **DQ проверки:** count, дубликаты (route_no, validity), NULL обязательных полей, ссылочная целостность (batch_id = текущий батч) +- **DQ проверки:** count, дубликаты (route_no, validity), NULL обязательных полей, ссылочная целостность (_load_id = текущий батч) #### stg.seats (справочник, full load) - **Источник:** `bookings.seats` (через PXF) @@ -173,9 +173,9 @@ - `airplane_code TEXT` - код самолёта - `seat_no TEXT` - номер места - `fare_conditions TEXT` - класс обслуживания -- **Технические колонки:** `src_created_at_ts`, `load_dttm`, `batch_id` +- **Технические колонки:** `event_ts` (=now()), `_load_ts`, `_load_id` - **Стратегия загрузки:** Full load -- **DQ проверки:** count, дубликаты (airplane_code, seat_no), NULL обязательных полей, ссылочная целостность (batch_id = текущий батч) +- **DQ проверки:** count, дубликаты (airplane_code, seat_no), NULL обязательных полей, ссылочная целостность (_load_id = текущий батч) #### stg.flights (транзакции, инкремент) - **Источник:** `bookings.flights` (через PXF) @@ -189,9 +189,9 @@ - `scheduled_arrival TEXT` - плановое время прилёта - `actual_departure TEXT` - фактическое время вылета - `actual_arrival TEXT` - фактическое время прилёта -- **Технические колонки:** `src_created_at_ts` (=scheduled_departure), `load_dttm`, `batch_id` +- **Технические колонки:** `event_ts` (=scheduled_departure), `_load_ts`, `_load_id` - **Стратегия загрузки:** Инкремент по `scheduled_departure` -- **DQ проверки:** count (окно инкремента, пустое окно допустимо), дубликаты flight_id, NULL обязательных полей, ссылочная целостность (routes, batch_id = текущий батч) +- **DQ проверки:** count (окно инкремента, пустое окно допустимо), дубликаты flight_id, NULL обязательных полей, ссылочная целостность (routes, _load_id = текущий батч) #### stg.segments (транзакции, инкремент) - **Источник:** `bookings.segments` (через PXF) @@ -202,7 +202,7 @@ - `flight_id TEXT` - идентификатор рейса - `fare_conditions TEXT` - класс обслуживания - `price TEXT` - цена -- **Технические колонки:** `src_created_at_ts` (из book_date через tickets), `load_dttm`, `batch_id` +- **Технические колонки:** `event_ts` (из book_date через tickets), `_load_ts`, `_load_id` - **Стратегия загрузки:** Инкремент по `book_date` (через tickets) - **DQ проверки:** count (окно инкремента, пустое окно допустимо), дубликаты (ticket_no, flight_id), NULL обязательных полей, ссылочная целостность (tickets, flights) @@ -216,7 +216,7 @@ - `seat_no TEXT` - номер места - `boarding_no TEXT` - номер посадки - `boarding_time TEXT` - время посадки -- **Технические колонки:** `src_created_at_ts` (=now()), `load_dttm`, `batch_id` +- **Технические колонки:** `event_ts` (=now()), `_load_ts`, `_load_id` - **Стратегия загрузки:** Full snapshot (все строки при каждом запуске) - **DQ проверки:** count, дубликаты (ticket_no, flight_id), NULL обязательных полей, ссылочная целостность diff --git a/docs/internal/naming_conventions.md b/docs/internal/naming_conventions.md index bba5590..ecabbf9 100644 --- a/docs/internal/naming_conventions.md +++ b/docs/internal/naming_conventions.md @@ -40,21 +40,17 @@ - `event_ts` (effective time) и `_load_ts` (load time) — разные сущности, не смешиваем. - Если `event_ts` отсутствует в источнике, используем `_load_ts` как fallback и явно документируем это в SQL/доке. +- Для snapshot-справочников (airports, airplanes, routes, seats) в STG нет бизнес-события с точным временем: `event_ts` заполняется через `now()` при загрузке. Это намеренно и задокументировано в каждом load-скрипте. ## 5. Применение по слоям ### STG -- Для уже реализованного `bookings` STG сохраняем текущие legacy-имена ради обратной совместимости: - - `src_created_at_ts` - - `load_dttm` - - `batch_id` -- Для новых STG-объектов (новые домены/задачи) используем канон `_load_id`, `_load_ts` (и `event_ts`, если нужно). +- Используем канон `_load_id`, `_load_ts`, `event_ts` для всех таблиц. ### ODS -- В новых реализациях используем канон: - - `_load_id`, `_load_ts`, `event_ts`. +- Используем канон `_load_id`, `_load_ts`, `event_ts`. - Базовый эталон ODS в этом стенде: SCD Type 1 (current state + UPSERT). ### DDS @@ -63,17 +59,7 @@ - `valid_from`, `valid_to`, `hashdiff`, `created_at`, `updated_at`. - Интервалы считаем как `[valid_from, valid_to)`, current-версия: `valid_to IS NULL`. -## 6. Переходный маппинг legacy -> канон - -| Legacy (текущий bookings STG) | Канон | -|---|---| -| `batch_id` | `_load_id` | -| `load_dttm` | `_load_ts` | -| `src_created_at_ts` | `event_ts` | - -Примечание: это логический маппинг для новых слоёв. Массовое переименование существующего STG не требуется. - -## 7. Что проверяем в ревью +## 6. Что проверяем в ревью - Нет новых техполей-синнонимов вроде `loaded_at`, `ingested_at`, `batch_key`, если уже есть канон. - Нет смешивания `event_ts` и `_load_ts` в одном смысле. diff --git a/docs/internal/qa-plan.md b/docs/internal/qa-plan.md index 2dca77d..8bf5c32 100644 --- a/docs/internal/qa-plan.md +++ b/docs/internal/qa-plan.md @@ -136,23 +136,22 @@ UNION ALL SELECT 'boarding_passes', ```sql -- Снапшот-таблицы: ODS = последний батч STG --- Примечание: поле называется batch_id (без подчёркивания), а не _batch_id SELECT 'airports' AS entity, (SELECT COUNT(*) FROM stg.airports - WHERE batch_id = (SELECT MAX(batch_id) FROM stg.airports)) AS stg_cnt, + WHERE _load_id = (SELECT MAX(_load_id) FROM stg.airports)) AS stg_cnt, (SELECT COUNT(*) FROM ods.airports) AS ods_cnt UNION ALL SELECT 'airplanes', (SELECT COUNT(*) FROM stg.airplanes - WHERE batch_id = (SELECT MAX(batch_id) FROM stg.airplanes)), + WHERE _load_id = (SELECT MAX(_load_id) FROM stg.airplanes)), (SELECT COUNT(*) FROM ods.airplanes) UNION ALL SELECT 'routes', (SELECT COUNT(*) FROM stg.routes - WHERE batch_id = (SELECT MAX(batch_id) FROM stg.routes)), + WHERE _load_id = (SELECT MAX(_load_id) FROM stg.routes)), (SELECT COUNT(*) FROM ods.routes) UNION ALL SELECT 'seats', (SELECT COUNT(*) FROM stg.seats - WHERE batch_id = (SELECT MAX(batch_id) FROM stg.seats)), + WHERE _load_id = (SELECT MAX(_load_id) FROM stg.seats)), (SELECT COUNT(*) FROM ods.seats); ``` @@ -281,7 +280,7 @@ WHERE f.calendar_sk IS NOT NULL AND c.calendar_sk IS NULL; ```sql SELECT COUNT(*) AS unprocessed FROM stg.bookings -WHERE load_dttm > (SELECT COALESCE(MAX(_load_ts), '1900-01-01') FROM ods.bookings); +WHERE _load_ts > (SELECT COALESCE(MAX(_load_ts), '1900-01-01') FROM ods.bookings); -- Ожидание: 0 ``` @@ -352,10 +351,10 @@ ORDER BY 1; #### B. STG хранит оба батча ```sql -SELECT _batch_id, COUNT(*) FROM stg.bookings GROUP BY 1 ORDER BY 1; +SELECT _load_id, COUNT(*) FROM stg.bookings GROUP BY 1 ORDER BY 1; ``` -**Ожидание:** 2 разных `_batch_id`, оба с данными. +**Ожидание:** 2 разных `_load_id`, оба с данными. #### C. Снапшоты не дублировались @@ -363,19 +362,19 @@ SELECT _batch_id, COUNT(*) FROM stg.bookings GROUP BY 1 ORDER BY 1; SELECT 'airports' AS entity, (SELECT COUNT(*) FROM stg.airports - WHERE _batch_id = (SELECT MAX(_batch_id) FROM stg.airports)) AS stg_last_batch, + WHERE _load_id = (SELECT MAX(_load_id) FROM stg.airports)) AS stg_last_batch, (SELECT COUNT(*) FROM ods.airports) AS ods_cnt UNION ALL SELECT 'airplanes', (SELECT COUNT(*) FROM stg.airplanes - WHERE _batch_id = (SELECT MAX(_batch_id) FROM stg.airplanes)), + WHERE _load_id = (SELECT MAX(_load_id) FROM stg.airplanes)), (SELECT COUNT(*) FROM ods.airplanes) UNION ALL SELECT 'routes', (SELECT COUNT(*) FROM stg.routes - WHERE _batch_id = (SELECT MAX(_batch_id) FROM stg.routes)), + WHERE _load_id = (SELECT MAX(_load_id) FROM stg.routes)), (SELECT COUNT(*) FROM ods.routes) UNION ALL SELECT 'seats', (SELECT COUNT(*) FROM stg.seats - WHERE _batch_id = (SELECT MAX(_batch_id) FROM stg.seats)), + WHERE _load_id = (SELECT MAX(_load_id) FROM stg.seats)), (SELECT COUNT(*) FROM ods.seats); ``` @@ -431,7 +430,7 @@ make bookings-generate-day #### A. Монотонный рост инкрементальных таблиц ```sql -SELECT _batch_id, COUNT(*) FROM stg.bookings GROUP BY 1 ORDER BY 1; +SELECT _load_id, COUNT(*) FROM stg.bookings GROUP BY 1 ORDER BY 1; ``` **Ожидание:** 5 строк, все с данными. @@ -483,25 +482,21 @@ SELECT **Цель:** проверить качество кода без запуска стенда. Можно делать параллельно с блоками 1-4. -### 5.1 Консистентность _load_id / _batch_id +### 5.1 Консистентность _load_id во всех слоях ```bash -# В STG должен быть _batch_id (через {{ run_id }}) -grep -r '_batch_id' sql/stg/*_load.sql | head -20 +# В STG/ODS/DDS/DM должен быть _load_id +grep -r '_load_id' sql/stg/*_load.sql sql/ods/*_load.sql sql/dds/*_load.sql sql/dm/*_load.sql | head -20 -# В ODS/DDS/DM должен быть _load_id (через {{ run_id }}) -grep -r '_load_id' sql/ods/*_load.sql sql/dds/*_load.sql sql/dm/*_load.sql | head -20 - -# НЕ должно быть: _load_id в STG или _batch_id в ODS/DDS/DM (кроме чтения из STG) -grep -r '_load_id' sql/stg/*_load.sql # ожидание: пусто -grep -r '_batch_id' sql/ods/*_load.sql # допустимо: чтение из stg -grep -r '_batch_id' sql/dds/*_load.sql sql/dm/*_load.sql # ожидание: пусто +# Не должно быть старых имён batch_id, load_dttm, src_created_at_ts +grep -r '\bbatch_id\b' sql/stg/ sql/ods/ sql/dds/ sql/dm/ # ожидание: только допустимые переменные PL (v_batch_id) +grep -r 'load_dttm\|src_created_at_ts' sql/ # ожидание: пусто ``` ### 5.2 Все load.sql используют шаблон {{ run_id }} Примечание: ODS намеренно не использует `{{ run_id }}` — вместо этого в `_load_id` -сохраняется `batch_id` из STG для сквозного lineage (traceable to source batch). +сохраняется `_load_id` из STG для сквозного lineage (traceable to source batch). Это правильный паттерн, а не баг. Проверять нужно только STG/DDS/DM. ```bash diff --git a/docs/internal/stg_naming_unification_plan.md b/docs/internal/stg_naming_unification_plan.md new file mode 100644 index 0000000..bca5fb9 --- /dev/null +++ b/docs/internal/stg_naming_unification_plan.md @@ -0,0 +1,118 @@ +# Унификация нейминга служебных полей в STG + +## Контекст + +В STG-слое используются legacy-имена служебных полей (`batch_id`, `load_dttm`, `src_created_at_ts`), а начиная с ODS — каноничные (`_load_id`, `_load_ts`, `event_ts`). Студент видит разные имена для одного понятия. Цель — привести STG к канону, убрав расхождение. + +> **Breaking change (dev-only).** Это ломающее переименование колонок. Миграционный шаг (ALTER TABLE … RENAME COLUMN) не предусмотрен. DDL-файлы используют `CREATE TABLE IF NOT EXISTS`, поэтому сами по себе они не пересоздадут существующие таблицы с новыми именами колонок. План предполагает заранее пересозданную среду (например, `make down && make up`) или ручной `DROP TABLE` / `DROP SCHEMA` перед `make ddl-gp`. Обратная совместимость не обеспечивается. + +## Маппинг + +| Legacy (STG сейчас) | Канон (ODS/DDS/DM) | +|---|---| +| `batch_id` | `_load_id` | +| `load_dttm` | `_load_ts` | +| `src_created_at_ts` | `event_ts` | + +### Оговорка про `event_ts` в snapshot-справочниках + +В транзакционных STG-таблицах (bookings, tickets, flights, segments, boarding_passes) поле `src_created_at_ts` действительно хранит время события из источника — переименование в `event_ts` семантически точно. + +В snapshot-справочниках (airports, airplanes, routes, seats) это поле заполняется `now()` при загрузке, т.е. по факту это ещё одно load-time, а не время события. Тем не менее мы сохраняем единое имя `event_ts` как **учебное упрощение** — ради консистентной структуры STG-таблиц. Это зафиксировано как осознанный trade-off: единообразие важнее семантической точности в справочниках. В `naming_conventions.md` нужно добавить соответствующую оговорку (раздел 4, «Time Rule»). + +## Что НЕ переименовываем + +- PL/pgSQL переменная `v_batch_id` — это локальная переменная, не колонка +- Python-функция `_resolve_stg_batch_id`, переменная `stg_batch_id`, task_id `resolve_stg_batch_id` — это Python/Airflow-идентификаторы +- XCom-ключи, ссылающиеся на task_id + +## Порядок выполнения + +### Шаг 1: STG DDL (9 файлов) + +`sql/stg/{bookings,tickets,flights,segments,airports,airplanes,routes,seats,boarding_passes}_ddl.sql` + +В каждом: `batch_id` → `_load_id`, `load_dttm` → `_load_ts`, `src_created_at_ts` → `event_ts`. + +### Шаг 2: STG Load (9 файлов) + +`sql/stg/{bookings,tickets,flights,segments,airports,airplanes,routes,seats,boarding_passes}_load.sql` + +INSERT-списки, SELECT, WHERE, комментарии — те же 3 замены. + +### Шаг 3: STG DQ (9 файлов) + +`sql/stg/{bookings,tickets,flights,segments,airports,airplanes,routes,seats,boarding_passes}_dq.sql` + +WHERE-условия (`batch_id = v_batch_id` → `_load_id = v_batch_id`), RAISE-сообщения, комментарии. + +### Шаг 4: ODS Load (9 файлов) + +Два подтипа — обрабатывать по-разному. + +#### 4a: Транзакционные таблицы (5 файлов) + +`sql/ods/{bookings,tickets,flights,segments,boarding_passes}_load.sql` + +SELECT из STG: `s.batch_id` → `s._load_id`, `s.load_dttm` → `s._load_ts`, `s.src_created_at_ts` → `s.event_ts`. Убрать лишние алиасы (`s.src_created_at_ts AS event_ts` → просто `s.event_ts`). + +#### 4b: Snapshot-справочники (4 файла) + +`sql/ods/{airports,airplanes,routes,seats}_load.sql` + +Здесь `event_ts` отсутствует в целевой ODS-таблице — менять только ссылки на STG-колонки: `s.batch_id` → `s._load_id`, `s.load_dttm` → `s._load_ts`, `s.src_created_at_ts` → `s.event_ts` (только в ORDER BY / WHERE, где они читают из STG). ODS-колонка `_load_ts` по-прежнему заполняется через `now()`, это не меняется. + +### Шаг 5: ODS DQ (9 файлов) + +`sql/ods/{bookings,tickets,flights,segments,airports,airplanes,routes,seats,boarding_passes}_dq.sql` + +`WHERE batch_id =` → `WHERE _load_id =`, RAISE-сообщения. + +### Шаг 6: DAG-файлы (2 файла) + +- `airflow/dags/bookings_to_gp_stage.py` — комментарий про `batch_id` +- `airflow/dags/bookings_to_gp_ods.py` — встроенный SQL-запрос резолвера: все `batch_id` как колонка → `_load_id`, `load_dttm` → `_load_ts`. Python-имена не трогаем. + +### Шаг 7: Тесты (3 файла) + +- `tests/test_ods_snapshot_integration.py` — inline DDL и INSERT в тестах +- `tests/test_dags_smoke.py` — комментарии +- `tests/test_ods_sql_contract.py` — docstring + +### Шаг 8: Документация (~15 файлов) + +- `docs/internal/naming_conventions.md` — убрать legacy-исключение (секция 5/STG), убрать переходный маппинг (секция 6), добавить оговорку про `event_ts` в snapshot-справочниках (секция 4) +- `docs/internal/db_schema.md` — описания STG-полей +- `docs/internal/bookings_stg_design.md` — дизайн STG +- `docs/internal/bookings_ods_design.md` — маппинг STG→ODS, SQL-примеры +- `docs/internal/qa-plan.md` — SQL-запросы проверок +- `docs/internal/architecture_review.md` — архитектурные заметки +- `docs/internal/bookings_stg_code_review.md` — код-ревью +- `docs/bookings_to_gp_stage.md` — описание STG DAG, примеры полей +- `docs/bookings_to_gp_ods.md` — описание ODS DAG +- `docs/dag_execution_order.md` — порядок выполнения DAG +- `educational-tasks.md` — учебные задания +- `README.md` — SQL-примеры в README +- `TESTING.md` — чек-лист тестирования + +### Шаг 9: Верификация + +```bash +# 1. Проверить SQL и Python — не должно быть колонок batch_id +# (допустимы только: v_batch_id, stg_batch_id, resolve_stg_batch_id) +grep -rn 'batch_id' sql/stg/ sql/ods/ airflow/dags/ tests/ --include='*.sql' --include='*.py' + +# 2. Ноль совпадений по старым именам в коде +grep -rn 'load_dttm' sql/ airflow/dags/ tests/ --include='*.sql' --include='*.py' +grep -rn 'src_created_at_ts' sql/ airflow/dags/ tests/ --include='*.sql' --include='*.py' + +# 3. Проверить документацию — не должно быть старых имён как актуальных +# (допустимы упоминания в историческом контексте) +grep -rn 'batch_id\|load_dttm\|src_created_at_ts' docs/ educational-tasks.md README.md TESTING.md + +# 4. Тесты и линтер +make test +make fmt && make lint +``` + +## Итого: ~55 файлов, ~3 механические замены в каждом diff --git a/educational-tasks.md b/educational-tasks.md index 031f956..1c78fd9 100644 --- a/educational-tasks.md +++ b/educational-tasks.md @@ -30,7 +30,7 @@ 1. Прочитайте `sql/stg/bookings_ddl.sql` и краткое описание потока `docs/bookings_to_gp_stage.md` (если интересно — `docs/internal/bookings_stg_design.md`). 2. Ответьте себе на вопросы: - чем внешняя таблица `stg.bookings_ext` отличается от внутренней `stg.bookings`; - - зачем нужны тех.колонки `src_created_at_ts`, `load_dttm`, `batch_id`; + - зачем нужны тех.колонки `event_ts`, `_load_ts`, `_load_id`; - чем слой STG отличается от итоговых витрин (DDS/DM) с точки зрения моделирования. 3. Выполните `make ddl-gp`, затем зайдите в Greenplum (`make gp-psql`) и проверьте наличие схемы и таблиц: - `\dn` и `\dt stg.*` @@ -71,7 +71,7 @@ - загрузка инкремента в `stg.bookings`; - проверка количества строк между источником и STG. 4. Обратите внимание, как в DAG используется логическая дата запуска: - - `{{ run_id }}` используется как `batch_id` — метка загрузки в таблице `stg.bookings` для конкретного запуска; + - `{{ run_id }}` используется как `_load_id` — метка загрузки в таблице `stg.bookings` для конкретного запуска; - сами даты данных (какие дни есть в `bookings.bookings`) определяются генератором по `max(book_date)`, а не по `ds`. На этом этапе достаточно понять общую цепочку. Детальные задания по переработке модели данных и построению ODS/DDS/DM слоёв будут добавлены позже. diff --git a/sql/ods/airplanes_dq.sql b/sql/ods/airplanes_dq.sql index 100c487..ee9b081 100644 --- a/sql/ods/airplanes_dq.sql +++ b/sql/ods/airplanes_dq.sql @@ -13,11 +13,11 @@ BEGIN SELECT COUNT(*) INTO v_stg_batch_count FROM stg.airplanes - WHERE batch_id = v_batch_id; + WHERE _load_id = v_batch_id; IF v_stg_batch_count = 0 THEN RAISE EXCEPTION - 'DQ FAILED: batch_id=% для stg.airplanes пустой. Проверьте загрузку STG и PXF.', + 'DQ FAILED: _load_id=% для stg.airplanes пустой. Проверьте загрузку STG и PXF.', v_batch_id; END IF; @@ -38,7 +38,7 @@ BEGIN FROM ( SELECT DISTINCT airplane_code FROM stg.airplanes - WHERE batch_id = v_batch_id + WHERE _load_id = v_batch_id ) AS s WHERE NOT EXISTS ( SELECT 1 @@ -48,7 +48,7 @@ BEGIN IF v_missing_keys_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: в ods.airplanes отсутствуют ключи из stg.airplanes (batch_id=%): %', + 'DQ FAILED: в ods.airplanes отсутствуют ключи из stg.airplanes (_load_id=%): %', v_batch_id, v_missing_keys_count; END IF; @@ -62,14 +62,14 @@ BEGIN FROM ( SELECT DISTINCT airplane_code FROM stg.airplanes - WHERE batch_id = v_batch_id + WHERE _load_id = v_batch_id ) AS s WHERE s.airplane_code = o.airplane_code ); IF v_extra_keys_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: в ods.airplanes найдены лишние ключи вне stg batch_id=%: %', + 'DQ FAILED: в ods.airplanes найдены лишние ключи вне stg _load_id=%: %', v_batch_id, v_extra_keys_count; END IF; @@ -93,7 +93,7 @@ BEGIN END IF; RAISE NOTICE - 'DQ PASSED: ods.airplanes ок (batch_id=%): stg_batch_rows=%', + 'DQ PASSED: ods.airplanes ок (_load_id=%): stg_batch_rows=%', v_batch_id, v_stg_batch_count; END $$; diff --git a/sql/ods/airplanes_load.sql b/sql/ods/airplanes_load.sql index ec7064a..48c1b19 100644 --- a/sql/ods/airplanes_load.sql +++ b/sql/ods/airplanes_load.sql @@ -21,10 +21,10 @@ WITH src AS ( NULLIF(s.speed, '')::INTEGER AS speed_kmh, ROW_NUMBER() OVER ( PARTITION BY s.airplane_code - ORDER BY s.load_dttm DESC, s.src_created_at_ts DESC NULLS LAST + ORDER BY s._load_ts DESC, s.event_ts DESC NULLS LAST ) AS rn FROM stg.airplanes AS s - WHERE s.batch_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text + WHERE s._load_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text ) SELECT s.airplane_code, diff --git a/sql/ods/airports_dq.sql b/sql/ods/airports_dq.sql index 46b3c56..b429e3e 100644 --- a/sql/ods/airports_dq.sql +++ b/sql/ods/airports_dq.sql @@ -13,11 +13,11 @@ BEGIN SELECT COUNT(*) INTO v_stg_batch_count FROM stg.airports - WHERE batch_id = v_batch_id; + WHERE _load_id = v_batch_id; IF v_stg_batch_count = 0 THEN RAISE EXCEPTION - 'DQ FAILED: batch_id=% для stg.airports пустой. Проверьте загрузку STG и PXF.', + 'DQ FAILED: _load_id=% для stg.airports пустой. Проверьте загрузку STG и PXF.', v_batch_id; END IF; @@ -38,7 +38,7 @@ BEGIN FROM ( SELECT DISTINCT airport_code FROM stg.airports - WHERE batch_id = v_batch_id + WHERE _load_id = v_batch_id ) AS s WHERE NOT EXISTS ( SELECT 1 @@ -48,7 +48,7 @@ BEGIN IF v_missing_keys_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: в ods.airports отсутствуют ключи из stg.airports (batch_id=%): %', + 'DQ FAILED: в ods.airports отсутствуют ключи из stg.airports (_load_id=%): %', v_batch_id, v_missing_keys_count; END IF; @@ -62,14 +62,14 @@ BEGIN FROM ( SELECT DISTINCT airport_code FROM stg.airports - WHERE batch_id = v_batch_id + WHERE _load_id = v_batch_id ) AS s WHERE s.airport_code = o.airport_code ); IF v_extra_keys_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: в ods.airports найдены лишние ключи вне stg batch_id=%: %', + 'DQ FAILED: в ods.airports найдены лишние ключи вне stg _load_id=%: %', v_batch_id, v_extra_keys_count; END IF; @@ -99,7 +99,7 @@ BEGIN END IF; RAISE NOTICE - 'DQ PASSED: ods.airports ок (batch_id=%): stg_batch_rows=%', + 'DQ PASSED: ods.airports ок (_load_id=%): stg_batch_rows=%', v_batch_id, v_stg_batch_count; END $$; diff --git a/sql/ods/airports_load.sql b/sql/ods/airports_load.sql index 6f11239..41470c7 100644 --- a/sql/ods/airports_load.sql +++ b/sql/ods/airports_load.sql @@ -25,10 +25,10 @@ WITH src AS ( s.timezone, ROW_NUMBER() OVER ( PARTITION BY s.airport_code - ORDER BY s.load_dttm DESC, s.src_created_at_ts DESC NULLS LAST + ORDER BY s._load_ts DESC, s.event_ts DESC NULLS LAST ) AS rn FROM stg.airports AS s - WHERE s.batch_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text + WHERE s._load_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text ) SELECT s.airport_code, diff --git a/sql/ods/boarding_passes_dq.sql b/sql/ods/boarding_passes_dq.sql index aeed460..5d30113 100644 --- a/sql/ods/boarding_passes_dq.sql +++ b/sql/ods/boarding_passes_dq.sql @@ -13,7 +13,7 @@ BEGIN SELECT COUNT(*) INTO v_stg_batch_count FROM stg.boarding_passes - WHERE batch_id = v_batch_id; + WHERE _load_id = v_batch_id; -- В ODS не должно быть дублей по составному бизнес-ключу. SELECT COUNT(*) @@ -37,7 +37,7 @@ BEGIN FROM ( SELECT DISTINCT ticket_no, NULLIF(flight_id, '')::INTEGER AS flight_id FROM stg.boarding_passes - WHERE batch_id = v_batch_id + WHERE _load_id = v_batch_id ) AS s WHERE NOT EXISTS ( SELECT 1 @@ -48,7 +48,7 @@ BEGIN IF v_missing_keys_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: в ods.boarding_passes отсутствуют ключи из stg.boarding_passes (batch_id=%): %', + 'DQ FAILED: в ods.boarding_passes отсутствуют ключи из stg.boarding_passes (_load_id=%): %', v_batch_id, v_missing_keys_count; END IF; @@ -90,7 +90,7 @@ BEGIN END IF; RAISE NOTICE - 'DQ PASSED: ods.boarding_passes ок (batch_id=%): stg_batch_rows=%', + 'DQ PASSED: ods.boarding_passes ок (_load_id=%): stg_batch_rows=%', v_batch_id, v_stg_batch_count; END $$; diff --git a/sql/ods/boarding_passes_load.sql b/sql/ods/boarding_passes_load.sql index 6d9754d..65fe336 100644 --- a/sql/ods/boarding_passes_load.sql +++ b/sql/ods/boarding_passes_load.sql @@ -10,16 +10,16 @@ WITH src AS ( s.seat_no, NULLIF(s.boarding_no, '')::INTEGER AS boarding_no, NULLIF(s.boarding_time, '')::TIMESTAMP WITH TIME ZONE AS boarding_time, - s.src_created_at_ts AS event_ts, - s.batch_id, - s.load_dttm, + s.event_ts, + s._load_id, + s._load_ts, ROW_NUMBER() OVER ( PARTITION BY s.ticket_no, s.flight_id - ORDER BY s.src_created_at_ts DESC NULLS LAST, s.load_dttm DESC + ORDER BY s.event_ts DESC NULLS LAST, s._load_ts DESC ) AS rn FROM stg.boarding_passes AS s -- Используем HWM (High Water Mark) по техническому времени STG - WHERE s.load_dttm > (SELECT COALESCE(MAX(_load_ts), '1900-01-01 00:00:00'::TIMESTAMP) FROM ods.boarding_passes) + WHERE s._load_ts > (SELECT COALESCE(MAX(_load_ts), '1900-01-01 00:00:00'::TIMESTAMP) FROM ods.boarding_passes) ) SELECT * FROM src WHERE rn = 1; @@ -29,8 +29,8 @@ SET seat_no = s.seat_no, boarding_no = s.boarding_no, boarding_time = s.boarding_time, event_ts = s.event_ts, - _load_id = s.batch_id, -- Сохраняем оригинальный lineage из STG - _load_ts = s.load_dttm -- Фиксируем время STG как водяной знак для ODS + _load_id = s._load_id, -- Сохраняем оригинальный lineage из STG + _load_ts = s._load_ts -- Фиксируем время STG как водяной знак для ODS FROM tmp_boarding_passes_delta AS s WHERE o.ticket_no = s.ticket_no AND o.flight_id = s.flight_id @@ -59,8 +59,8 @@ SELECT s.boarding_no, s.boarding_time, s.event_ts, - s.batch_id, - s.load_dttm + s._load_id, + s._load_ts FROM tmp_boarding_passes_delta AS s WHERE NOT EXISTS ( SELECT 1 diff --git a/sql/ods/bookings_dq.sql b/sql/ods/bookings_dq.sql index 1709423..3425cb0 100644 --- a/sql/ods/bookings_dq.sql +++ b/sql/ods/bookings_dq.sql @@ -12,7 +12,7 @@ BEGIN SELECT COUNT(*) INTO v_stg_batch_count FROM stg.bookings - WHERE batch_id = v_batch_id; + WHERE _load_id = v_batch_id; -- В ODS не должно быть дублей по бизнес-ключу. SELECT COUNT(*) - COUNT(DISTINCT book_ref) @@ -31,7 +31,7 @@ BEGIN FROM ( SELECT DISTINCT book_ref FROM stg.bookings - WHERE batch_id = v_batch_id + WHERE _load_id = v_batch_id ) AS s WHERE NOT EXISTS ( SELECT 1 @@ -41,7 +41,7 @@ BEGIN IF v_missing_keys_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: в ods.bookings отсутствуют ключи из stg.bookings (batch_id=%): %', + 'DQ FAILED: в ods.bookings отсутствуют ключи из stg.bookings (_load_id=%): %', v_batch_id, v_missing_keys_count; END IF; @@ -65,7 +65,7 @@ BEGIN END IF; RAISE NOTICE - 'DQ PASSED: ods.bookings ок (batch_id=%): stg_batch_rows=%', + 'DQ PASSED: ods.bookings ок (_load_id=%): stg_batch_rows=%', v_batch_id, v_stg_batch_count; END $$; diff --git a/sql/ods/bookings_load.sql b/sql/ods/bookings_load.sql index e796c40..74f5a01 100644 --- a/sql/ods/bookings_load.sql +++ b/sql/ods/bookings_load.sql @@ -8,16 +8,16 @@ WITH src AS ( s.book_ref, NULLIF(s.book_date, '')::TIMESTAMP WITH TIME ZONE AS book_date, NULLIF(s.total_amount, '')::NUMERIC(10,2) AS total_amount, - s.src_created_at_ts AS event_ts, - s.batch_id, - s.load_dttm, + s.event_ts, + s._load_id, + s._load_ts, ROW_NUMBER() OVER ( PARTITION BY s.book_ref - ORDER BY s.src_created_at_ts DESC NULLS LAST, s.load_dttm DESC + ORDER BY s.event_ts DESC NULLS LAST, s._load_ts DESC ) AS rn FROM stg.bookings AS s -- Используем HWM (High Water Mark) по техническому времени STG - WHERE s.load_dttm > (SELECT COALESCE(MAX(_load_ts), '1900-01-01 00:00:00'::TIMESTAMP) FROM ods.bookings) + WHERE s._load_ts > (SELECT COALESCE(MAX(_load_ts), '1900-01-01 00:00:00'::TIMESTAMP) FROM ods.bookings) ) SELECT * FROM src WHERE rn = 1; @@ -26,8 +26,8 @@ UPDATE ods.bookings AS o SET book_date = s.book_date, total_amount = s.total_amount, event_ts = s.event_ts, - _load_id = s.batch_id, -- Сохраняем оригинальный lineage из STG - _load_ts = s.load_dttm -- Фиксируем время STG как водяной знак для ODS + _load_id = s._load_id, -- Сохраняем оригинальный lineage из STG + _load_ts = s._load_ts -- Фиксируем время STG как водяной знак для ODS FROM tmp_bookings_delta AS s WHERE o.book_ref = s.book_ref AND ( @@ -50,8 +50,8 @@ SELECT s.book_date, s.total_amount, s.event_ts, - s.batch_id, - s.load_dttm + s._load_id, + s._load_ts FROM tmp_bookings_delta AS s WHERE NOT EXISTS ( SELECT 1 diff --git a/sql/ods/flights_dq.sql b/sql/ods/flights_dq.sql index 912e340..0a400df 100644 --- a/sql/ods/flights_dq.sql +++ b/sql/ods/flights_dq.sql @@ -13,7 +13,7 @@ BEGIN SELECT COUNT(*) INTO v_stg_batch_count FROM stg.flights - WHERE batch_id = v_batch_id; + WHERE _load_id = v_batch_id; -- В ODS не должно быть дублей по бизнес-ключу. SELECT COUNT(*) - COUNT(DISTINCT flight_id) @@ -32,7 +32,7 @@ BEGIN FROM ( SELECT DISTINCT NULLIF(flight_id, '')::INTEGER AS flight_id FROM stg.flights - WHERE batch_id = v_batch_id + WHERE _load_id = v_batch_id ) AS s WHERE NOT EXISTS ( SELECT 1 @@ -42,7 +42,7 @@ BEGIN IF v_missing_keys_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: в ods.flights отсутствуют ключи из stg.flights (batch_id=%): %', + 'DQ FAILED: в ods.flights отсутствуют ключи из stg.flights (_load_id=%): %', v_batch_id, v_missing_keys_count; END IF; @@ -83,7 +83,7 @@ BEGIN END IF; RAISE NOTICE - 'DQ PASSED: ods.flights ок (batch_id=%): stg_batch_rows=%', + 'DQ PASSED: ods.flights ок (_load_id=%): stg_batch_rows=%', v_batch_id, v_stg_batch_count; END $$; diff --git a/sql/ods/flights_load.sql b/sql/ods/flights_load.sql index 02db1e8..f2bb5a5 100644 --- a/sql/ods/flights_load.sql +++ b/sql/ods/flights_load.sql @@ -9,7 +9,7 @@ WITH segment_flights AS ( SELECT DISTINCT NULLIF(s.flight_id, '')::INTEGER AS flight_id FROM stg.segments AS s - WHERE s.load_dttm > (SELECT COALESCE(MAX(_load_ts), '1900-01-01 00:00:00'::TIMESTAMP) FROM ods.segments) + WHERE s._load_ts > (SELECT COALESCE(MAX(_load_ts), '1900-01-01 00:00:00'::TIMESTAMP) FROM ods.segments) AND s.flight_id IS NOT NULL AND s.flight_id <> '' ), @@ -22,9 +22,9 @@ stg_flights_typed AS ( NULLIF(s.scheduled_arrival, '')::TIMESTAMP WITH TIME ZONE AS scheduled_arrival, NULLIF(s.actual_departure, '')::TIMESTAMP WITH TIME ZONE AS actual_departure, NULLIF(s.actual_arrival, '')::TIMESTAMP WITH TIME ZONE AS actual_arrival, - s.src_created_at_ts AS event_ts, - s.load_dttm, - s.batch_id + s.event_ts, + s._load_ts, + s._load_id FROM stg.flights AS s WHERE s.flight_id IS NOT NULL AND s.flight_id <> '' @@ -40,10 +40,10 @@ src_union AS ( f.actual_departure, f.actual_arrival, f.event_ts, - f.load_dttm, - f.batch_id + f._load_ts, + f._load_id FROM stg_flights_typed AS f - WHERE f.load_dttm > (SELECT COALESCE(MAX(_load_ts), '1900-01-01 00:00:00'::TIMESTAMP) FROM ods.flights) + WHERE f._load_ts > (SELECT COALESCE(MAX(_load_ts), '1900-01-01 00:00:00'::TIMESTAMP) FROM ods.flights) UNION ALL @@ -57,8 +57,8 @@ src_union AS ( f.actual_departure, f.actual_arrival, f.event_ts, - f.load_dttm, - f.batch_id + f._load_ts, + f._load_id FROM stg_flights_typed AS f JOIN segment_flights AS sf ON sf.flight_id = f.flight_id @@ -73,11 +73,11 @@ src AS ( u.actual_departure, u.actual_arrival, u.event_ts, - u.batch_id, - u.load_dttm, + u._load_id, + u._load_ts, ROW_NUMBER() OVER ( PARTITION BY u.flight_id - ORDER BY u.event_ts DESC NULLS LAST, u.load_dttm DESC + ORDER BY u.event_ts DESC NULLS LAST, u._load_ts DESC ) AS rn FROM src_union AS u ) @@ -92,8 +92,8 @@ SET route_no = s.route_no, actual_departure = s.actual_departure, actual_arrival = s.actual_arrival, event_ts = s.event_ts, - _load_id = s.batch_id, -- Сохраняем оригинальный lineage из STG - _load_ts = s.load_dttm -- Фиксируем время STG как водяной знак для ODS + _load_id = s._load_id, -- Сохраняем оригинальный lineage из STG + _load_ts = s._load_ts -- Фиксируем время STG как водяной знак для ODS FROM tmp_flights_delta AS s WHERE o.flight_id = s.flight_id AND ( @@ -128,8 +128,8 @@ SELECT s.actual_departure, s.actual_arrival, s.event_ts, - s.batch_id, - s.load_dttm + s._load_id, + s._load_ts FROM tmp_flights_delta AS s WHERE NOT EXISTS ( SELECT 1 diff --git a/sql/ods/routes_dq.sql b/sql/ods/routes_dq.sql index 2b36969..4dcf325 100644 --- a/sql/ods/routes_dq.sql +++ b/sql/ods/routes_dq.sql @@ -16,11 +16,11 @@ BEGIN SELECT COUNT(*) INTO v_stg_batch_count FROM stg.routes - WHERE batch_id = v_batch_id; + WHERE _load_id = v_batch_id; IF v_stg_batch_count = 0 THEN RAISE EXCEPTION - 'DQ FAILED: batch_id=% для stg.routes пустой. Проверьте загрузку STG и PXF.', + 'DQ FAILED: _load_id=% для stg.routes пустой. Проверьте загрузку STG и PXF.', v_batch_id; END IF; @@ -46,7 +46,7 @@ BEGIN FROM ( SELECT DISTINCT route_no, validity FROM stg.routes - WHERE batch_id = v_batch_id + WHERE _load_id = v_batch_id ) AS s WHERE NOT EXISTS ( SELECT 1 @@ -57,7 +57,7 @@ BEGIN IF v_missing_keys_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: в ods.routes отсутствуют ключи из stg.routes (batch_id=%): %', + 'DQ FAILED: в ods.routes отсутствуют ключи из stg.routes (_load_id=%): %', v_batch_id, v_missing_keys_count; END IF; @@ -71,7 +71,7 @@ BEGIN FROM ( SELECT DISTINCT route_no, validity FROM stg.routes - WHERE batch_id = v_batch_id + WHERE _load_id = v_batch_id ) AS s WHERE s.route_no = o.route_no AND s.validity = o.validity @@ -79,7 +79,7 @@ BEGIN IF v_extra_keys_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: в ods.routes найдены лишние ключи вне stg batch_id=%: %', + 'DQ FAILED: в ods.routes найдены лишние ключи вне stg _load_id=%: %', v_batch_id, v_extra_keys_count; END IF; @@ -157,7 +157,7 @@ BEGIN END IF; RAISE NOTICE - 'DQ PASSED: ods.routes ок (batch_id=%): stg_batch_rows=%', + 'DQ PASSED: ods.routes ок (_load_id=%): stg_batch_rows=%', v_batch_id, v_stg_batch_count; END $$; diff --git a/sql/ods/routes_load.sql b/sql/ods/routes_load.sql index f692494..81b3af5 100644 --- a/sql/ods/routes_load.sql +++ b/sql/ods/routes_load.sql @@ -29,10 +29,10 @@ WITH src AS ( NULLIF(s.duration, '')::INTERVAL AS duration, ROW_NUMBER() OVER ( PARTITION BY s.route_no, s.validity - ORDER BY s.load_dttm DESC, s.src_created_at_ts DESC NULLS LAST + ORDER BY s._load_ts DESC, s.event_ts DESC NULLS LAST ) AS rn FROM stg.routes AS s - WHERE s.batch_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text + WHERE s._load_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text ) SELECT s.route_no, diff --git a/sql/ods/seats_dq.sql b/sql/ods/seats_dq.sql index ea2b61e..e8c4111 100644 --- a/sql/ods/seats_dq.sql +++ b/sql/ods/seats_dq.sql @@ -14,11 +14,11 @@ BEGIN SELECT COUNT(*) INTO v_stg_batch_count FROM stg.seats - WHERE batch_id = v_batch_id; + WHERE _load_id = v_batch_id; IF v_stg_batch_count = 0 THEN RAISE EXCEPTION - 'DQ FAILED: batch_id=% для stg.seats пустой. Проверьте загрузку STG и PXF.', + 'DQ FAILED: _load_id=% для stg.seats пустой. Проверьте загрузку STG и PXF.', v_batch_id; END IF; @@ -44,7 +44,7 @@ BEGIN FROM ( SELECT DISTINCT airplane_code, seat_no FROM stg.seats - WHERE batch_id = v_batch_id + WHERE _load_id = v_batch_id ) AS s WHERE NOT EXISTS ( SELECT 1 @@ -55,7 +55,7 @@ BEGIN IF v_missing_keys_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: в ods.seats отсутствуют ключи из stg.seats (batch_id=%): %', + 'DQ FAILED: в ods.seats отсутствуют ключи из stg.seats (_load_id=%): %', v_batch_id, v_missing_keys_count; END IF; @@ -69,7 +69,7 @@ BEGIN FROM ( SELECT DISTINCT airplane_code, seat_no FROM stg.seats - WHERE batch_id = v_batch_id + WHERE _load_id = v_batch_id ) AS s WHERE s.airplane_code = o.airplane_code AND s.seat_no = o.seat_no @@ -77,7 +77,7 @@ BEGIN IF v_extra_keys_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: в ods.seats найдены лишние ключи вне stg batch_id=%: %', + 'DQ FAILED: в ods.seats найдены лишние ключи вне stg _load_id=%: %', v_batch_id, v_extra_keys_count; END IF; @@ -119,7 +119,7 @@ BEGIN END IF; RAISE NOTICE - 'DQ PASSED: ods.seats ок (batch_id=%): stg_batch_rows=%', + 'DQ PASSED: ods.seats ок (_load_id=%): stg_batch_rows=%', v_batch_id, v_stg_batch_count; END $$; diff --git a/sql/ods/seats_load.sql b/sql/ods/seats_load.sql index 7544958..4d3d091 100644 --- a/sql/ods/seats_load.sql +++ b/sql/ods/seats_load.sql @@ -19,10 +19,10 @@ WITH src AS ( s.fare_conditions, ROW_NUMBER() OVER ( PARTITION BY s.airplane_code, s.seat_no - ORDER BY s.load_dttm DESC, s.src_created_at_ts DESC NULLS LAST + ORDER BY s._load_ts DESC, s.event_ts DESC NULLS LAST ) AS rn FROM stg.seats AS s - WHERE s.batch_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text + WHERE s._load_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text ) SELECT s.airplane_code, diff --git a/sql/ods/segments_dq.sql b/sql/ods/segments_dq.sql index f1a05bc..a2e8403 100644 --- a/sql/ods/segments_dq.sql +++ b/sql/ods/segments_dq.sql @@ -14,7 +14,7 @@ BEGIN SELECT COUNT(*) INTO v_stg_batch_count FROM stg.segments - WHERE batch_id = v_batch_id; + WHERE _load_id = v_batch_id; -- В ODS не должно быть дублей по составному бизнес-ключу. SELECT COUNT(*) @@ -38,7 +38,7 @@ BEGIN FROM ( SELECT DISTINCT ticket_no, NULLIF(flight_id, '')::INTEGER AS flight_id FROM stg.segments - WHERE batch_id = v_batch_id + WHERE _load_id = v_batch_id ) AS s WHERE NOT EXISTS ( SELECT 1 @@ -49,7 +49,7 @@ BEGIN IF v_missing_keys_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: в ods.segments отсутствуют ключи из stg.segments (batch_id=%): %', + 'DQ FAILED: в ods.segments отсутствуют ключи из stg.segments (_load_id=%): %', v_batch_id, v_missing_keys_count; END IF; @@ -106,7 +106,7 @@ BEGIN END IF; RAISE NOTICE - 'DQ PASSED: ods.segments ок (batch_id=%): stg_batch_rows=%', + 'DQ PASSED: ods.segments ок (_load_id=%): stg_batch_rows=%', v_batch_id, v_stg_batch_count; END $$; diff --git a/sql/ods/segments_load.sql b/sql/ods/segments_load.sql index e6aa2d2..661cc8d 100644 --- a/sql/ods/segments_load.sql +++ b/sql/ods/segments_load.sql @@ -9,16 +9,16 @@ WITH src AS ( NULLIF(s.flight_id, '')::INTEGER AS flight_id, s.fare_conditions, NULLIF(s.price, '')::NUMERIC(10,2) AS amount, - s.src_created_at_ts AS event_ts, - s.batch_id, - s.load_dttm, + s.event_ts, + s._load_id, + s._load_ts, ROW_NUMBER() OVER ( PARTITION BY s.ticket_no, s.flight_id - ORDER BY s.src_created_at_ts DESC NULLS LAST, s.load_dttm DESC + ORDER BY s.event_ts DESC NULLS LAST, s._load_ts DESC ) AS rn FROM stg.segments AS s -- Используем HWM (High Water Mark) по техническому времени STG - WHERE s.load_dttm > (SELECT COALESCE(MAX(_load_ts), '1900-01-01 00:00:00'::TIMESTAMP) FROM ods.segments) + WHERE s._load_ts > (SELECT COALESCE(MAX(_load_ts), '1900-01-01 00:00:00'::TIMESTAMP) FROM ods.segments) ) SELECT * FROM src WHERE rn = 1; @@ -27,8 +27,8 @@ UPDATE ods.segments AS o SET fare_conditions = s.fare_conditions, amount = s.amount, event_ts = s.event_ts, - _load_id = s.batch_id, -- Сохраняем оригинальный lineage из STG - _load_ts = s.load_dttm -- Фиксируем время STG как водяной знак для ODS + _load_id = s._load_id, -- Сохраняем оригинальный lineage из STG + _load_ts = s._load_ts -- Фиксируем время STG как водяной знак для ODS FROM tmp_segments_delta AS s WHERE o.ticket_no = s.ticket_no AND o.flight_id = s.flight_id @@ -54,8 +54,8 @@ SELECT s.fare_conditions, s.amount, s.event_ts, - s.batch_id, - s.load_dttm + s._load_id, + s._load_ts FROM tmp_segments_delta AS s WHERE NOT EXISTS ( SELECT 1 diff --git a/sql/ods/tickets_dq.sql b/sql/ods/tickets_dq.sql index 3ad1007..638784b 100644 --- a/sql/ods/tickets_dq.sql +++ b/sql/ods/tickets_dq.sql @@ -13,7 +13,7 @@ BEGIN SELECT COUNT(*) INTO v_stg_batch_count FROM stg.tickets - WHERE batch_id = v_batch_id; + WHERE _load_id = v_batch_id; -- В ODS не должно быть дублей по бизнес-ключу. SELECT COUNT(*) - COUNT(DISTINCT ticket_no) @@ -32,7 +32,7 @@ BEGIN FROM ( SELECT DISTINCT ticket_no FROM stg.tickets - WHERE batch_id = v_batch_id + WHERE _load_id = v_batch_id ) AS s WHERE NOT EXISTS ( SELECT 1 @@ -42,7 +42,7 @@ BEGIN IF v_missing_keys_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: в ods.tickets отсутствуют ключи из stg.tickets (batch_id=%): %', + 'DQ FAILED: в ods.tickets отсутствуют ключи из stg.tickets (_load_id=%): %', v_batch_id, v_missing_keys_count; END IF; @@ -86,7 +86,7 @@ BEGIN END IF; RAISE NOTICE - 'DQ PASSED: ods.tickets ок (batch_id=%): stg_batch_rows=%', + 'DQ PASSED: ods.tickets ок (_load_id=%): stg_batch_rows=%', v_batch_id, v_stg_batch_count; END $$; diff --git a/sql/ods/tickets_load.sql b/sql/ods/tickets_load.sql index 4f03bf2..c0bf4c3 100644 --- a/sql/ods/tickets_load.sql +++ b/sql/ods/tickets_load.sql @@ -10,16 +10,16 @@ WITH src AS ( s.passenger_id, s.passenger_name, NULLIF(s.outbound, '')::BOOLEAN AS is_outbound, - s.src_created_at_ts AS event_ts, - s.batch_id, - s.load_dttm, + s.event_ts, + s._load_id, + s._load_ts, ROW_NUMBER() OVER ( PARTITION BY s.ticket_no - ORDER BY s.src_created_at_ts DESC NULLS LAST, s.load_dttm DESC + ORDER BY s.event_ts DESC NULLS LAST, s._load_ts DESC ) AS rn FROM stg.tickets AS s -- Используем HWM (High Water Mark) по техническому времени STG - WHERE s.load_dttm > (SELECT COALESCE(MAX(_load_ts), '1900-01-01 00:00:00'::TIMESTAMP) FROM ods.tickets) + WHERE s._load_ts > (SELECT COALESCE(MAX(_load_ts), '1900-01-01 00:00:00'::TIMESTAMP) FROM ods.tickets) ) SELECT * FROM src WHERE rn = 1; @@ -30,8 +30,8 @@ SET book_ref = s.book_ref, passenger_name = s.passenger_name, is_outbound = s.is_outbound, event_ts = s.event_ts, - _load_id = s.batch_id, -- Сохраняем оригинальный lineage из STG - _load_ts = s.load_dttm -- Фиксируем время STG как водяной знак для ODS + _load_id = s._load_id, -- Сохраняем оригинальный lineage из STG + _load_ts = s._load_ts -- Фиксируем время STG как водяной знак для ODS FROM tmp_tickets_delta AS s WHERE o.ticket_no = s.ticket_no AND ( @@ -60,8 +60,8 @@ SELECT s.passenger_name, s.is_outbound, s.event_ts, - s.batch_id, - s.load_dttm + s._load_id, + s._load_ts FROM tmp_tickets_delta AS s WHERE NOT EXISTS ( SELECT 1 diff --git a/sql/stg/airplanes_ddl.sql b/sql/stg/airplanes_ddl.sql index f252983..46aa28b 100644 --- a/sql/stg/airplanes_ddl.sql +++ b/sql/stg/airplanes_ddl.sql @@ -23,9 +23,9 @@ CREATE TABLE IF NOT EXISTS stg.airplanes ( model TEXT, range TEXT, speed TEXT, - src_created_at_ts TIMESTAMP, - load_dttm TIMESTAMP NOT NULL DEFAULT now(), - batch_id TEXT + event_ts TIMESTAMP, + _load_ts TIMESTAMP NOT NULL DEFAULT now(), + _load_id TEXT NOT NULL ) WITH (appendonly=true, orientation=row, compresstype=zstd, compresslevel=1) -- Ключ распределения: airplane_code diff --git a/sql/stg/airplanes_dq.sql b/sql/stg/airplanes_dq.sql index becc26f..e05b260 100644 --- a/sql/stg/airplanes_dq.sql +++ b/sql/stg/airplanes_dq.sql @@ -22,7 +22,7 @@ BEGIN SELECT COUNT(*) INTO v_stg_count FROM stg.airplanes - WHERE batch_id = v_batch_id; + WHERE _load_id = v_batch_id; IF v_src_count <> v_stg_count THEN RAISE EXCEPTION @@ -35,11 +35,11 @@ BEGIN SELECT COUNT(*) - COUNT(DISTINCT airplane_code) INTO v_dup_count FROM stg.airplanes AS a - WHERE a.batch_id = v_batch_id; + WHERE a._load_id = v_batch_id; IF v_dup_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены дубликаты airplane_code (batch_id=%): %', + 'DQ FAILED: найдены дубликаты airplane_code (_load_id=%): %', v_batch_id, v_dup_count; END IF; @@ -48,19 +48,19 @@ BEGIN SELECT COUNT(*) INTO v_null_count FROM stg.airplanes AS a - WHERE a.batch_id = v_batch_id + WHERE a._load_id = v_batch_id AND (a.airplane_code IS NULL OR a.airplane_code = '' OR a.model IS NULL OR a.model = ''); IF v_null_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены строки с NULL в обязательных полях (airplane_code, model) (batch_id=%): %', + 'DQ FAILED: найдены строки с NULL в обязательных полях (airplane_code, model) (_load_id=%): %', v_batch_id, v_null_count; END IF; RAISE NOTICE - 'DQ PASSED: airplanes ок (batch_id=%): source=% stg=%', + 'DQ PASSED: airplanes ок (_load_id=%): source=% stg=%', v_batch_id, v_src_count, v_stg_count; diff --git a/sql/stg/airplanes_load.sql b/sql/stg/airplanes_load.sql index 49d7157..b846b9c 100644 --- a/sql/stg/airplanes_load.sql +++ b/sql/stg/airplanes_load.sql @@ -1,14 +1,14 @@ -- Загрузка всех строк из stg.airplanes_ext в stg.airplanes (full load). --- Используем batch_id для отслеживания загрузки. +-- Используем _load_id для отслеживания загрузки. INSERT INTO stg.airplanes ( airplane_code, model, range, speed, - src_created_at_ts, - load_dttm, - batch_id + event_ts, + _load_ts, + _load_id ) SELECT ext.airplane_code::text, @@ -20,10 +20,10 @@ SELECT '{{ run_id }}'::text FROM stg.airplanes_ext AS ext WHERE NOT EXISTS ( - -- Идемпотентность: при повторном запуске/ретрае не вставляем повторно те же строки в рамках текущего batch_id. + -- Идемпотентность: при повторном запуске/ретрае не вставляем повторно те же строки в рамках текущего _load_id. SELECT 1 FROM stg.airplanes AS a - WHERE a.batch_id = '{{ run_id }}'::text + WHERE a._load_id = '{{ run_id }}'::text AND a.airplane_code = ext.airplane_code::text ); diff --git a/sql/stg/airports_ddl.sql b/sql/stg/airports_ddl.sql index f35a281..708478f 100644 --- a/sql/stg/airports_ddl.sql +++ b/sql/stg/airports_ddl.sql @@ -27,9 +27,9 @@ CREATE TABLE IF NOT EXISTS stg.airports ( country TEXT, coordinates TEXT, timezone TEXT, - src_created_at_ts TIMESTAMP, - load_dttm TIMESTAMP NOT NULL DEFAULT now(), - batch_id TEXT + event_ts TIMESTAMP, + _load_ts TIMESTAMP NOT NULL DEFAULT now(), + _load_id TEXT NOT NULL ) WITH (appendonly=true, orientation=row, compresstype=zstd, compresslevel=1) -- Ключ распределения: airport_code diff --git a/sql/stg/airports_dq.sql b/sql/stg/airports_dq.sql index 3de90e2..8d49bf3 100644 --- a/sql/stg/airports_dq.sql +++ b/sql/stg/airports_dq.sql @@ -22,7 +22,7 @@ BEGIN SELECT COUNT(*) INTO v_stg_count FROM stg.airports - WHERE batch_id = v_batch_id; + WHERE _load_id = v_batch_id; IF v_src_count <> v_stg_count THEN RAISE EXCEPTION @@ -35,11 +35,11 @@ BEGIN SELECT COUNT(*) - COUNT(DISTINCT airport_code) INTO v_dup_count FROM stg.airports AS a - WHERE a.batch_id = v_batch_id; + WHERE a._load_id = v_batch_id; IF v_dup_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены дубликаты airport_code (batch_id=%): %', + 'DQ FAILED: найдены дубликаты airport_code (_load_id=%): %', v_batch_id, v_dup_count; END IF; @@ -48,7 +48,7 @@ BEGIN SELECT COUNT(*) INTO v_null_count FROM stg.airports AS a - WHERE a.batch_id = v_batch_id + WHERE a._load_id = v_batch_id AND (a.airport_code IS NULL OR a.airport_code = '' OR a.airport_name IS NULL OR a.airport_name = '' OR a.city IS NULL OR a.city = '' @@ -56,13 +56,13 @@ BEGIN IF v_null_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены строки с NULL в обязательных полях (airport_code, airport_name, city, timezone) (batch_id=%): %', + 'DQ FAILED: найдены строки с NULL в обязательных полях (airport_code, airport_name, city, timezone) (_load_id=%): %', v_batch_id, v_null_count; END IF; RAISE NOTICE - 'DQ PASSED: airports ок (batch_id=%): source=% stg=%', + 'DQ PASSED: airports ок (_load_id=%): source=% stg=%', v_batch_id, v_src_count, v_stg_count; diff --git a/sql/stg/airports_load.sql b/sql/stg/airports_load.sql index a3bb1a1..c42b142 100644 --- a/sql/stg/airports_load.sql +++ b/sql/stg/airports_load.sql @@ -1,5 +1,5 @@ -- Загрузка всех строк из stg.airports_ext в stg.airports (full load). --- Используем batch_id для отслеживания загрузки. +-- Используем _load_id для отслеживания загрузки. INSERT INTO stg.airports ( airport_code, @@ -8,9 +8,9 @@ INSERT INTO stg.airports ( country, coordinates, timezone, - src_created_at_ts, - load_dttm, - batch_id + event_ts, + _load_ts, + _load_id ) SELECT ext.airport_code::text, @@ -24,10 +24,10 @@ SELECT '{{ run_id }}'::text FROM stg.airports_ext AS ext WHERE NOT EXISTS ( - -- Идемпотентность: при повторном запуске/ретрае не вставляем повторно те же строки в рамках текущего batch_id. + -- Идемпотентность: при повторном запуске/ретрае не вставляем повторно те же строки в рамках текущего _load_id. SELECT 1 FROM stg.airports AS a - WHERE a.batch_id = '{{ run_id }}'::text + WHERE a._load_id = '{{ run_id }}'::text AND a.airport_code = ext.airport_code::text ); diff --git a/sql/stg/boarding_passes_ddl.sql b/sql/stg/boarding_passes_ddl.sql index 5b432be..29669c8 100644 --- a/sql/stg/boarding_passes_ddl.sql +++ b/sql/stg/boarding_passes_ddl.sql @@ -24,9 +24,9 @@ CREATE TABLE IF NOT EXISTS stg.boarding_passes ( seat_no TEXT, boarding_no TEXT, boarding_time TEXT, - src_created_at_ts TIMESTAMP, - load_dttm TIMESTAMP NOT NULL DEFAULT now(), - batch_id TEXT + event_ts TIMESTAMP, + _load_ts TIMESTAMP NOT NULL DEFAULT now(), + _load_id TEXT NOT NULL ) WITH (appendonly=true, orientation=row, compresstype=zstd, compresslevel=1) -- Ключ распределения: ticket_no diff --git a/sql/stg/boarding_passes_dq.sql b/sql/stg/boarding_passes_dq.sql index 6fcf686..e6cd7a2 100644 --- a/sql/stg/boarding_passes_dq.sql +++ b/sql/stg/boarding_passes_dq.sql @@ -17,7 +17,7 @@ BEGIN IF v_src_count = 0 THEN RAISE NOTICE - 'В источнике boarding_passes_ext нет строк - пропускаем DQ проверки (batch_id=%).', + 'В источнике boarding_passes_ext нет строк - пропускаем DQ проверки (_load_id=%).', v_batch_id; RETURN; END IF; @@ -26,7 +26,7 @@ BEGIN SELECT COUNT(*) INTO v_stg_count FROM stg.boarding_passes - WHERE batch_id = v_batch_id; + WHERE _load_id = v_batch_id; IF v_src_count <> v_stg_count THEN RAISE EXCEPTION @@ -40,11 +40,11 @@ BEGIN SELECT COUNT(*) - COUNT(DISTINCT md5(ROW(ticket_no, flight_id)::text)) INTO v_dup_count FROM stg.boarding_passes AS bp - WHERE bp.batch_id = v_batch_id; + WHERE bp._load_id = v_batch_id; IF v_dup_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены дубликаты (ticket_no, flight_id) (batch_id=%): %', + 'DQ FAILED: найдены дубликаты (ticket_no, flight_id) (_load_id=%): %', v_batch_id, v_dup_count; END IF; @@ -53,13 +53,13 @@ BEGIN SELECT COUNT(*) INTO v_null_count FROM stg.boarding_passes AS bp - WHERE bp.batch_id = v_batch_id + WHERE bp._load_id = v_batch_id AND (bp.ticket_no IS NULL OR bp.ticket_no = '' OR bp.flight_id IS NULL OR bp.flight_id = ''); IF v_null_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены строки с NULL в обязательных полях (batch_id=%): %', + 'DQ FAILED: найдены строки с NULL в обязательных полях (_load_id=%): %', v_batch_id, v_null_count; END IF; @@ -69,12 +69,12 @@ BEGIN INTO v_orphan_ticket_count FROM stg.boarding_passes AS bp LEFT JOIN stg.tickets AS t ON bp.ticket_no = t.ticket_no - WHERE bp.batch_id = v_batch_id + WHERE bp._load_id = v_batch_id AND t.ticket_no IS NULL; IF v_orphan_ticket_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены boarding_passes без соответствующих tickets (batch_id=%): %', + 'DQ FAILED: найдены boarding_passes без соответствующих tickets (_load_id=%): %', v_batch_id, v_orphan_ticket_count; END IF; @@ -84,23 +84,23 @@ BEGIN INTO v_orphan_segment_count FROM stg.boarding_passes AS bp LEFT JOIN stg.segments AS s ON bp.ticket_no = s.ticket_no AND bp.flight_id = s.flight_id - WHERE bp.batch_id = v_batch_id + WHERE bp._load_id = v_batch_id AND s.ticket_no IS NULL; IF v_orphan_segment_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены boarding_passes без соответствующих segments (batch_id=%): %', + 'DQ FAILED: найдены boarding_passes без соответствующих segments (_load_id=%): %', v_batch_id, v_orphan_segment_count; END IF; RAISE NOTICE - 'DQ PASSED: boarding_passes ок (batch_id=%): source=% stg=%', + 'DQ PASSED: boarding_passes ок (_load_id=%): source=% stg=%', v_batch_id, v_src_count, v_stg_count; EXCEPTION WHEN OTHERS THEN - RAISE NOTICE 'DQ ERROR для boarding_passes (batch_id=%): %', v_batch_id, SQLERRM; + RAISE NOTICE 'DQ ERROR для boarding_passes (_load_id=%): %', v_batch_id, SQLERRM; RAISE; END $$; diff --git a/sql/stg/boarding_passes_load.sql b/sql/stg/boarding_passes_load.sql index 00899db..8afe640 100644 --- a/sql/stg/boarding_passes_load.sql +++ b/sql/stg/boarding_passes_load.sql @@ -1,6 +1,6 @@ -- Загрузка всех строк из stg.boarding_passes_ext в stg.boarding_passes. -- Используем full snapshot: все строки при каждом запуске. --- Используем batch_id для отслеживания загрузки. +-- Используем _load_id для отслеживания загрузки. INSERT INTO stg.boarding_passes ( ticket_no, @@ -8,9 +8,9 @@ INSERT INTO stg.boarding_passes ( seat_no, boarding_no, boarding_time, - src_created_at_ts, - load_dttm, - batch_id + event_ts, + _load_ts, + _load_id ) SELECT ext.ticket_no, @@ -23,11 +23,11 @@ SELECT '{{ run_id }}'::text FROM stg.boarding_passes_ext AS ext WHERE NOT EXISTS ( - -- Идемпотентность: при повторном запуске/ретрае не вставляем повторно те же строки в рамках текущего batch_id. + -- Идемпотентность: при повторном запуске/ретрае не вставляем повторно те же строки в рамках текущего _load_id. -- Считаем ключом строки (ticket_no, flight_id). SELECT 1 FROM stg.boarding_passes AS bp - WHERE bp.batch_id = '{{ run_id }}'::text + WHERE bp._load_id = '{{ run_id }}'::text AND bp.ticket_no = ext.ticket_no AND bp.flight_id = ext.flight_id ); diff --git a/sql/stg/bookings_ddl.sql b/sql/stg/bookings_ddl.sql index 6c04f1e..13b860f 100644 --- a/sql/stg/bookings_ddl.sql +++ b/sql/stg/bookings_ddl.sql @@ -20,9 +20,9 @@ CREATE TABLE IF NOT EXISTS stg.bookings ( book_ref TEXT, book_date TEXT, total_amount TEXT, - src_created_at_ts TIMESTAMP, - load_dttm TIMESTAMP NOT NULL DEFAULT now(), - batch_id TEXT NOT NULL + event_ts TIMESTAMP, + _load_ts TIMESTAMP NOT NULL DEFAULT now(), + _load_id TEXT NOT NULL ) WITH (appendonly=true, orientation=row, compresstype=zstd, compresslevel=1) -- Ключ распределения: book_ref diff --git a/sql/stg/bookings_dq.sql b/sql/stg/bookings_dq.sql index 65c8f79..b8c7f7d 100644 --- a/sql/stg/bookings_dq.sql +++ b/sql/stg/bookings_dq.sql @@ -1,7 +1,7 @@ -- Проверка количества строк между источником stg.bookings_ext и стейджем stg.bookings. -- Считаем строки за то же окно инкремента, что и при загрузке: --- все записи во внешней таблице с book_date больше максимального src_created_at_ts --- из предыдущих батчей должны совпасть по количеству со строками текущего batch_id. +-- все записи во внешней таблице с book_date больше максимального event_ts +-- из предыдущих батчей должны совпасть по количеству со строками текущего _load_id. DO $$ DECLARE @@ -12,12 +12,12 @@ DECLARE v_dup_count bigint; v_null_amount_count bigint; BEGIN - -- Опорная метка: максимум src_created_at_ts среди предыдущих батчей - SELECT max(src_created_at_ts) + -- Опорная метка: максимум event_ts среди предыдущих батчей + SELECT max(event_ts) INTO v_prev_ts FROM stg.bookings - WHERE batch_id <> v_batch_id - OR batch_id IS NULL; + WHERE _load_id <> v_batch_id + OR _load_id IS NULL; -- Источник: считаем строки во внешней таблице, которые вошли в новое окно SELECT COUNT(*) @@ -27,21 +27,21 @@ BEGIN IF v_src_count = 0 THEN -- Пустое окно инкремента допустимо: новых данных может не быть. - -- В этом случае ожидаем, что в текущем batch_id тоже 0 строк. + -- В этом случае ожидаем, что в текущем _load_id тоже 0 строк. SELECT COUNT(*) INTO v_stg_count FROM stg.bookings - WHERE batch_id = v_batch_id; + WHERE _load_id = v_batch_id; IF v_stg_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: источник bookings_ext за окно инкремента пустой, но в stg.bookings есть строки текущего batch_id (batch_id=%): %', + 'DQ FAILED: источник bookings_ext за окно инкремента пустой, но в stg.bookings есть строки текущего _load_id (_load_id=%): %', v_batch_id, v_stg_count; END IF; RAISE NOTICE - 'В источнике bookings_ext нет строк для окна инкремента (book_date > %). Пропускаем DQ проверки (batch_id=%).', + 'В источнике bookings_ext нет строк для окна инкремента (book_date > %). Пропускаем DQ проверки (_load_id=%).', COALESCE(v_prev_ts, TIMESTAMP '1900-01-01 00:00:00'), v_batch_id; RETURN; @@ -51,7 +51,7 @@ BEGIN SELECT COUNT(*) INTO v_stg_count FROM stg.bookings - WHERE batch_id = v_batch_id; + WHERE _load_id = v_batch_id; IF v_src_count <> v_stg_count THEN RAISE EXCEPTION @@ -64,11 +64,11 @@ BEGIN SELECT COUNT(*) - COUNT(DISTINCT book_ref) INTO v_dup_count FROM stg.bookings AS b - WHERE b.batch_id = v_batch_id; + WHERE b._load_id = v_batch_id; IF v_dup_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены дубликаты book_ref (batch_id=%): %', + 'DQ FAILED: найдены дубликаты book_ref (_load_id=%): %', v_batch_id, v_dup_count; END IF; @@ -77,12 +77,12 @@ BEGIN SELECT COUNT(*) INTO v_null_amount_count FROM stg.bookings AS b - WHERE b.batch_id = v_batch_id + WHERE b._load_id = v_batch_id AND (b.total_amount IS NULL OR b.total_amount = ''); IF v_null_amount_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены bookings с NULL или пустым total_amount (batch_id=%): %', + 'DQ FAILED: найдены bookings с NULL или пустым total_amount (_load_id=%): %', v_batch_id, v_null_amount_count; END IF; diff --git a/sql/stg/bookings_load.sql b/sql/stg/bookings_load.sql index 6c336a4..c22b618 100644 --- a/sql/stg/bookings_load.sql +++ b/sql/stg/bookings_load.sql @@ -1,22 +1,22 @@ -- Загрузка инкремента из stg.bookings_ext в stg.bookings. --- Окно инкремента определяется по src_created_at_ts: --- берём строки, где book_date больше максимального src_created_at_ts +-- Окно инкремента определяется по event_ts: +-- берём строки, где book_date больше максимального event_ts -- среди "старых" батчей; верхняя граница по дате не используется. -- CTE для определения максимальной даты загрузки предыдущего батча WITH max_batch_ts AS ( - SELECT COALESCE(MAX(src_created_at_ts), TIMESTAMP '1900-01-01 00:00:00') AS max_ts + SELECT COALESCE(MAX(event_ts), TIMESTAMP '1900-01-01 00:00:00') AS max_ts FROM stg.bookings - WHERE batch_id <> '{{ run_id }}'::text - OR batch_id IS NULL + WHERE _load_id <> '{{ run_id }}'::text + OR _load_id IS NULL ) INSERT INTO stg.bookings ( book_ref, book_date, total_amount, - src_created_at_ts, - load_dttm, - batch_id + event_ts, + _load_ts, + _load_id ) SELECT ext.book_ref::text, @@ -29,10 +29,10 @@ FROM stg.bookings_ext AS ext CROSS JOIN max_batch_ts AS mb WHERE ext.book_date > mb.max_ts AND NOT EXISTS ( - -- Идемпотентность: при повторном запуске/ретрае не вставляем повторно те же строки в рамках текущего batch_id. + -- Идемпотентность: при повторном запуске/ретрае не вставляем повторно те же строки в рамках текущего _load_id. SELECT 1 FROM stg.bookings AS b - WHERE b.batch_id = '{{ run_id }}'::text + WHERE b._load_id = '{{ run_id }}'::text AND b.book_ref = ext.book_ref::text ); diff --git a/sql/stg/flights_ddl.sql b/sql/stg/flights_ddl.sql index 7a6d623..cffd5a1 100644 --- a/sql/stg/flights_ddl.sql +++ b/sql/stg/flights_ddl.sql @@ -28,9 +28,9 @@ CREATE TABLE IF NOT EXISTS stg.flights ( scheduled_arrival TEXT, actual_departure TEXT, actual_arrival TEXT, - src_created_at_ts TIMESTAMP, - load_dttm TIMESTAMP NOT NULL DEFAULT now(), - batch_id TEXT + event_ts TIMESTAMP, + _load_ts TIMESTAMP NOT NULL DEFAULT now(), + _load_id TEXT NOT NULL ) WITH (appendonly=true, orientation=row, compresstype=zstd, compresslevel=1) -- Ключ распределения: flight_id diff --git a/sql/stg/flights_dq.sql b/sql/stg/flights_dq.sql index 97a0b72..bdeba8c 100644 --- a/sql/stg/flights_dq.sql +++ b/sql/stg/flights_dq.sql @@ -10,12 +10,12 @@ DECLARE v_null_count BIGINT; v_orphan_route_count BIGINT; BEGIN - -- Опорная метка: максимум src_created_at_ts среди предыдущих батчей - SELECT max(src_created_at_ts) + -- Опорная метка: максимум event_ts среди предыдущих батчей + SELECT max(event_ts) INTO v_prev_ts FROM stg.flights - WHERE batch_id <> v_batch_id - OR batch_id IS NULL; + WHERE _load_id <> v_batch_id + OR _load_id IS NULL; -- Источник: считаем строки во внешней таблице, которые вошли в окно инкремента SELECT COUNT(*) @@ -25,21 +25,21 @@ BEGIN IF v_src_count = 0 THEN -- Пустое окно инкремента допустимо: новых данных может не быть. - -- В этом случае ожидаем, что в текущем batch_id тоже 0 строк. + -- В этом случае ожидаем, что в текущем _load_id тоже 0 строк. SELECT COUNT(*) INTO v_stg_count FROM stg.flights - WHERE batch_id = v_batch_id; + WHERE _load_id = v_batch_id; IF v_stg_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: источник flights_ext за окно инкремента пустой, но в stg.flights есть строки текущего batch_id (batch_id=%): %', + 'DQ FAILED: источник flights_ext за окно инкремента пустой, но в stg.flights есть строки текущего _load_id (_load_id=%): %', v_batch_id, v_stg_count; END IF; RAISE NOTICE - 'В источнике flights_ext нет строк для окна инкремента (scheduled_departure > %). Пропускаем DQ проверки (batch_id=%).', + 'В источнике flights_ext нет строк для окна инкремента (scheduled_departure > %). Пропускаем DQ проверки (_load_id=%).', COALESCE(v_prev_ts, TIMESTAMP '1900-01-01 00:00:00'), v_batch_id; RETURN; @@ -49,7 +49,7 @@ BEGIN SELECT COUNT(*) INTO v_stg_count FROM stg.flights - WHERE batch_id = v_batch_id; + WHERE _load_id = v_batch_id; IF v_src_count <> v_stg_count THEN RAISE EXCEPTION @@ -62,11 +62,11 @@ BEGIN SELECT COUNT(*) - COUNT(DISTINCT flight_id) INTO v_dup_count FROM stg.flights AS f - WHERE f.batch_id = v_batch_id; + WHERE f._load_id = v_batch_id; IF v_dup_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены дубликаты flight_id (batch_id=%): %', + 'DQ FAILED: найдены дубликаты flight_id (_load_id=%): %', v_batch_id, v_dup_count; END IF; @@ -75,7 +75,7 @@ BEGIN SELECT COUNT(*) INTO v_null_count FROM stg.flights AS f - WHERE f.batch_id = v_batch_id + WHERE f._load_id = v_batch_id AND (f.flight_id IS NULL OR f.flight_id = '' OR f.route_no IS NULL OR f.route_no = '' OR f.status IS NULL OR f.status = '' @@ -83,7 +83,7 @@ BEGIN IF v_null_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены строки с NULL в обязательных полях (batch_id=%): %', + 'DQ FAILED: найдены строки с NULL в обязательных полях (_load_id=%): %', v_batch_id, v_null_count; END IF; @@ -94,19 +94,19 @@ BEGIN FROM stg.flights AS f LEFT JOIN stg.routes AS r ON f.route_no = r.route_no - AND r.batch_id = v_batch_id - WHERE f.batch_id = v_batch_id + AND r._load_id = v_batch_id + WHERE f._load_id = v_batch_id AND r.route_no IS NULL; IF v_orphan_route_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены flights без соответствующих routes (batch_id=%): %', + 'DQ FAILED: найдены flights без соответствующих routes (_load_id=%): %', v_batch_id, v_orphan_route_count; END IF; RAISE NOTICE - 'DQ PASSED: flights ок (batch_id=%): source=% stg=%', + 'DQ PASSED: flights ок (_load_id=%): source=% stg=%', v_batch_id, v_src_count, v_stg_count; diff --git a/sql/stg/flights_load.sql b/sql/stg/flights_load.sql index e7879dd..d3a47c3 100644 --- a/sql/stg/flights_load.sql +++ b/sql/stg/flights_load.sql @@ -1,14 +1,14 @@ -- Загрузка инкремента из stg.flights_ext в stg.flights. --- Окно инкремента определяется по src_created_at_ts: --- берём строки, где scheduled_departure больше максимального src_created_at_ts +-- Окно инкремента определяется по event_ts: +-- берём строки, где scheduled_departure больше максимального event_ts -- среди "старых" батчей; верхняя граница по дате не используется. -- CTE для определения максимальной даты загрузки предыдущего батча WITH max_batch_ts AS ( - SELECT COALESCE(MAX(src_created_at_ts), TIMESTAMP '1900-01-01 00:00:00') AS max_ts + SELECT COALESCE(MAX(event_ts), TIMESTAMP '1900-01-01 00:00:00') AS max_ts FROM stg.flights - WHERE batch_id <> '{{ run_id }}'::text - OR batch_id IS NULL + WHERE _load_id <> '{{ run_id }}'::text + OR _load_id IS NULL ) INSERT INTO stg.flights ( flight_id, @@ -18,9 +18,9 @@ INSERT INTO stg.flights ( scheduled_arrival, actual_departure, actual_arrival, - src_created_at_ts, - load_dttm, - batch_id + event_ts, + _load_ts, + _load_id ) SELECT ext.flight_id::text, @@ -37,10 +37,10 @@ FROM stg.flights_ext AS ext CROSS JOIN max_batch_ts AS mb WHERE ext.scheduled_departure > mb.max_ts AND NOT EXISTS ( - -- Идемпотентность: при повторном запуске/ретрае не вставляем повторно те же строки в рамках текущего batch_id. + -- Идемпотентность: при повторном запуске/ретрае не вставляем повторно те же строки в рамках текущего _load_id. SELECT 1 FROM stg.flights AS f - WHERE f.batch_id = '{{ run_id }}'::text + WHERE f._load_id = '{{ run_id }}'::text AND f.flight_id = ext.flight_id::text ); diff --git a/sql/stg/routes_ddl.sql b/sql/stg/routes_ddl.sql index f10a78e..2570196 100644 --- a/sql/stg/routes_ddl.sql +++ b/sql/stg/routes_ddl.sql @@ -31,9 +31,9 @@ CREATE TABLE IF NOT EXISTS stg.routes ( days_of_week TEXT, scheduled_time TEXT, duration TEXT, - src_created_at_ts TIMESTAMP, - load_dttm TIMESTAMP NOT NULL DEFAULT now(), - batch_id TEXT + event_ts TIMESTAMP, + _load_ts TIMESTAMP NOT NULL DEFAULT now(), + _load_id TEXT NOT NULL ) WITH (appendonly=true, orientation=row, compresstype=zstd, compresslevel=1) -- Ключ распределения: route_no diff --git a/sql/stg/routes_dq.sql b/sql/stg/routes_dq.sql index 80af52a..0a924a7 100644 --- a/sql/stg/routes_dq.sql +++ b/sql/stg/routes_dq.sql @@ -24,7 +24,7 @@ BEGIN SELECT COUNT(*) INTO v_stg_count FROM stg.routes - WHERE batch_id = v_batch_id; + WHERE _load_id = v_batch_id; IF v_src_count <> v_stg_count THEN RAISE EXCEPTION @@ -38,11 +38,11 @@ BEGIN SELECT COUNT(*) - COUNT(DISTINCT md5(ROW(route_no, validity)::text)) INTO v_dup_count FROM stg.routes AS r - WHERE r.batch_id = v_batch_id; + WHERE r._load_id = v_batch_id; IF v_dup_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены дубликаты (route_no, validity) (batch_id=%): %', + 'DQ FAILED: найдены дубликаты (route_no, validity) (_load_id=%): %', v_batch_id, v_dup_count; END IF; @@ -51,7 +51,7 @@ BEGIN SELECT COUNT(*) INTO v_null_count FROM stg.routes AS r - WHERE r.batch_id = v_batch_id + WHERE r._load_id = v_batch_id AND (r.route_no IS NULL OR r.route_no = '' OR r.departure_airport IS NULL OR r.departure_airport = '' OR r.arrival_airport IS NULL OR r.arrival_airport = '' @@ -59,7 +59,7 @@ BEGIN IF v_null_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены строки с NULL в обязательных полях (route_no, departure_airport, arrival_airport, airplane_code) (batch_id=%): %', + 'DQ FAILED: найдены строки с NULL в обязательных полях (route_no, departure_airport, arrival_airport, airplane_code) (_load_id=%): %', v_batch_id, v_null_count; END IF; @@ -70,13 +70,13 @@ BEGIN FROM stg.routes AS r LEFT JOIN stg.airports AS da ON r.departure_airport = da.airport_code - AND da.batch_id = v_batch_id - WHERE r.batch_id = v_batch_id + AND da._load_id = v_batch_id + WHERE r._load_id = v_batch_id AND da.airport_code IS NULL; IF v_orphan_airports_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены routes с несуществующим departure_airport в airports (batch_id=%): %', + 'DQ FAILED: найдены routes с несуществующим departure_airport в airports (_load_id=%): %', v_batch_id, v_orphan_airports_count; END IF; @@ -87,13 +87,13 @@ BEGIN FROM stg.routes AS r LEFT JOIN stg.airports AS aa ON r.arrival_airport = aa.airport_code - AND aa.batch_id = v_batch_id - WHERE r.batch_id = v_batch_id + AND aa._load_id = v_batch_id + WHERE r._load_id = v_batch_id AND aa.airport_code IS NULL; IF v_orphan_airports_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены routes с несуществующим arrival_airport в airports (batch_id=%): %', + 'DQ FAILED: найдены routes с несуществующим arrival_airport в airports (_load_id=%): %', v_batch_id, v_orphan_airports_count; END IF; @@ -104,19 +104,19 @@ BEGIN FROM stg.routes AS r LEFT JOIN stg.airplanes AS a ON r.airplane_code = a.airplane_code - AND a.batch_id = v_batch_id - WHERE r.batch_id = v_batch_id + AND a._load_id = v_batch_id + WHERE r._load_id = v_batch_id AND a.airplane_code IS NULL; IF v_orphan_airplanes_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены routes с несуществующим airplane_code в airplanes (batch_id=%): %', + 'DQ FAILED: найдены routes с несуществующим airplane_code в airplanes (_load_id=%): %', v_batch_id, v_orphan_airplanes_count; END IF; RAISE NOTICE - 'DQ PASSED: routes ок (batch_id=%): source=% stg=%', + 'DQ PASSED: routes ок (_load_id=%): source=% stg=%', v_batch_id, v_src_count, v_stg_count; diff --git a/sql/stg/routes_load.sql b/sql/stg/routes_load.sql index 1bb76a1..fb16869 100644 --- a/sql/stg/routes_load.sql +++ b/sql/stg/routes_load.sql @@ -1,5 +1,5 @@ -- Загрузка всех строк из stg.routes_ext в stg.routes (full load). --- Используем batch_id для отслеживания загрузки. +-- Используем _load_id для отслеживания загрузки. INSERT INTO stg.routes ( route_no, @@ -10,9 +10,9 @@ INSERT INTO stg.routes ( days_of_week, scheduled_time, duration, - src_created_at_ts, - load_dttm, - batch_id + event_ts, + _load_ts, + _load_id ) SELECT ext.route_no::text, @@ -28,11 +28,11 @@ SELECT '{{ run_id }}'::text FROM stg.routes_ext AS ext WHERE NOT EXISTS ( - -- Идемпотентность: при повторном запуске/ретрае не вставляем повторно те же строки в рамках текущего batch_id. + -- Идемпотентность: при повторном запуске/ретрае не вставляем повторно те же строки в рамках текущего _load_id. -- Считаем ключом строки (route_no, validity). SELECT 1 FROM stg.routes AS r - WHERE r.batch_id = '{{ run_id }}'::text + WHERE r._load_id = '{{ run_id }}'::text AND r.route_no = ext.route_no::text AND r.validity = ext.validity::text ); diff --git a/sql/stg/seats_ddl.sql b/sql/stg/seats_ddl.sql index ae81bc4..b1f6fa9 100644 --- a/sql/stg/seats_ddl.sql +++ b/sql/stg/seats_ddl.sql @@ -20,9 +20,9 @@ CREATE TABLE IF NOT EXISTS stg.seats ( airplane_code TEXT, seat_no TEXT, fare_conditions TEXT, - src_created_at_ts TIMESTAMP, - load_dttm TIMESTAMP NOT NULL DEFAULT now(), - batch_id TEXT + event_ts TIMESTAMP, + _load_ts TIMESTAMP NOT NULL DEFAULT now(), + _load_id TEXT NOT NULL ) WITH (appendonly=true, orientation=row, compresstype=zstd, compresslevel=1) -- Ключ распределения: airplane_code diff --git a/sql/stg/seats_dq.sql b/sql/stg/seats_dq.sql index 54744c9..86e55cc 100644 --- a/sql/stg/seats_dq.sql +++ b/sql/stg/seats_dq.sql @@ -23,7 +23,7 @@ BEGIN SELECT COUNT(*) INTO v_stg_count FROM stg.seats - WHERE batch_id = v_batch_id; + WHERE _load_id = v_batch_id; IF v_src_count <> v_stg_count THEN RAISE EXCEPTION @@ -37,11 +37,11 @@ BEGIN SELECT COUNT(*) - COUNT(DISTINCT md5(ROW(airplane_code, seat_no)::text)) INTO v_dup_count FROM stg.seats AS s - WHERE s.batch_id = v_batch_id; + WHERE s._load_id = v_batch_id; IF v_dup_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены дубликаты (airplane_code, seat_no) (batch_id=%): %', + 'DQ FAILED: найдены дубликаты (airplane_code, seat_no) (_load_id=%): %', v_batch_id, v_dup_count; END IF; @@ -50,14 +50,14 @@ BEGIN SELECT COUNT(*) INTO v_null_count FROM stg.seats AS s - WHERE s.batch_id = v_batch_id + WHERE s._load_id = v_batch_id AND (s.airplane_code IS NULL OR s.airplane_code = '' OR s.seat_no IS NULL OR s.seat_no = '' OR s.fare_conditions IS NULL OR s.fare_conditions = ''); IF v_null_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены строки с NULL в обязательных полях (airplane_code, seat_no, fare_conditions) (batch_id=%): %', + 'DQ FAILED: найдены строки с NULL в обязательных полях (airplane_code, seat_no, fare_conditions) (_load_id=%): %', v_batch_id, v_null_count; END IF; @@ -68,19 +68,19 @@ BEGIN FROM stg.seats AS s LEFT JOIN stg.airplanes AS a ON s.airplane_code = a.airplane_code - AND a.batch_id = v_batch_id - WHERE s.batch_id = v_batch_id + AND a._load_id = v_batch_id + WHERE s._load_id = v_batch_id AND a.airplane_code IS NULL; IF v_orphan_airplanes_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены seats с несуществующим airplane_code в airplanes (batch_id=%): %', + 'DQ FAILED: найдены seats с несуществующим airplane_code в airplanes (_load_id=%): %', v_batch_id, v_orphan_airplanes_count; END IF; RAISE NOTICE - 'DQ PASSED: seats ок (batch_id=%): source=% stg=%', + 'DQ PASSED: seats ок (_load_id=%): source=% stg=%', v_batch_id, v_src_count, v_stg_count; diff --git a/sql/stg/seats_load.sql b/sql/stg/seats_load.sql index 2e9720b..3e53b63 100644 --- a/sql/stg/seats_load.sql +++ b/sql/stg/seats_load.sql @@ -1,13 +1,13 @@ -- Загрузка всех строк из stg.seats_ext в stg.seats (full load). --- Используем batch_id для отслеживания загрузки. +-- Используем _load_id для отслеживания загрузки. INSERT INTO stg.seats ( airplane_code, seat_no, fare_conditions, - src_created_at_ts, - load_dttm, - batch_id + event_ts, + _load_ts, + _load_id ) SELECT ext.airplane_code::text, @@ -18,11 +18,11 @@ SELECT '{{ run_id }}'::text FROM stg.seats_ext AS ext WHERE NOT EXISTS ( - -- Идемпотентность: при повторном запуске/ретрае не вставляем повторно те же строки в рамках текущего batch_id. + -- Идемпотентность: при повторном запуске/ретрае не вставляем повторно те же строки в рамках текущего _load_id. -- Считаем ключом строки (airplane_code, seat_no). SELECT 1 FROM stg.seats AS s - WHERE s.batch_id = '{{ run_id }}'::text + WHERE s._load_id = '{{ run_id }}'::text AND s.airplane_code = ext.airplane_code::text AND s.seat_no = ext.seat_no::text ); diff --git a/sql/stg/segments_ddl.sql b/sql/stg/segments_ddl.sql index 1d5273f..aaae53a 100644 --- a/sql/stg/segments_ddl.sql +++ b/sql/stg/segments_ddl.sql @@ -22,9 +22,9 @@ CREATE TABLE IF NOT EXISTS stg.segments ( flight_id TEXT, fare_conditions TEXT, price TEXT, - src_created_at_ts TIMESTAMP, - load_dttm TIMESTAMP NOT NULL DEFAULT now(), - batch_id TEXT + event_ts TIMESTAMP, + _load_ts TIMESTAMP NOT NULL DEFAULT now(), + _load_id TEXT NOT NULL ) WITH (appendonly=true, orientation=row, compresstype=zstd, compresslevel=1) -- Ключ распределения: ticket_no diff --git a/sql/stg/segments_dq.sql b/sql/stg/segments_dq.sql index feb0aa1..78a7dc2 100644 --- a/sql/stg/segments_dq.sql +++ b/sql/stg/segments_dq.sql @@ -11,12 +11,12 @@ DECLARE v_orphan_ticket_count BIGINT; v_orphan_flight_count BIGINT; BEGIN - -- Опорная метка: максимум src_created_at_ts среди предыдущих батчей - SELECT max(src_created_at_ts) + -- Опорная метка: максимум event_ts среди предыдущих батчей + SELECT max(event_ts) INTO v_prev_ts FROM stg.segments - WHERE batch_id <> v_batch_id - OR batch_id IS NULL; + WHERE _load_id <> v_batch_id + OR _load_id IS NULL; -- Источник: считаем строки во внешней таблице, которые вошли в окно инкремента SELECT COUNT(*) @@ -28,21 +28,21 @@ BEGIN IF v_src_count = 0 THEN -- Пустое окно инкремента допустимо: новых данных может не быть. - -- В этом случае ожидаем, что в текущем batch_id тоже 0 строк. + -- В этом случае ожидаем, что в текущем _load_id тоже 0 строк. SELECT COUNT(*) INTO v_stg_count FROM stg.segments - WHERE batch_id = v_batch_id; + WHERE _load_id = v_batch_id; IF v_stg_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: источник segments_ext за окно инкремента пустой, но в stg.segments есть строки текущего batch_id (batch_id=%): %', + 'DQ FAILED: источник segments_ext за окно инкремента пустой, но в stg.segments есть строки текущего _load_id (_load_id=%): %', v_batch_id, v_stg_count; END IF; RAISE NOTICE - 'В источнике segments_ext нет строк для окна инкремента (book_date > %). Пропускаем DQ проверки (batch_id=%).', + 'В источнике segments_ext нет строк для окна инкремента (book_date > %). Пропускаем DQ проверки (_load_id=%).', COALESCE(v_prev_ts, TIMESTAMP '1900-01-01 00:00:00'), v_batch_id; RETURN; @@ -52,7 +52,7 @@ BEGIN SELECT COUNT(*) INTO v_stg_count FROM stg.segments - WHERE batch_id = v_batch_id; + WHERE _load_id = v_batch_id; IF v_src_count <> v_stg_count THEN RAISE EXCEPTION @@ -66,11 +66,11 @@ BEGIN SELECT COUNT(*) - COUNT(DISTINCT md5(ROW(ticket_no, flight_id)::text)) INTO v_dup_count FROM stg.segments AS s - WHERE s.batch_id = v_batch_id; + WHERE s._load_id = v_batch_id; IF v_dup_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены дубликаты (ticket_no, flight_id) (batch_id=%): %', + 'DQ FAILED: найдены дубликаты (ticket_no, flight_id) (_load_id=%): %', v_batch_id, v_dup_count; END IF; @@ -79,7 +79,7 @@ BEGIN SELECT COUNT(*) INTO v_null_count FROM stg.segments AS s - WHERE s.batch_id = v_batch_id + WHERE s._load_id = v_batch_id AND (s.ticket_no IS NULL OR s.ticket_no = '' OR s.flight_id IS NULL OR s.flight_id = '' OR s.fare_conditions IS NULL OR s.fare_conditions = '' @@ -87,7 +87,7 @@ BEGIN IF v_null_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены строки с NULL в обязательных полях (batch_id=%): %', + 'DQ FAILED: найдены строки с NULL в обязательных полях (_load_id=%): %', v_batch_id, v_null_count; END IF; @@ -97,12 +97,12 @@ BEGIN INTO v_orphan_ticket_count FROM stg.segments AS s LEFT JOIN stg.tickets AS t ON s.ticket_no = t.ticket_no - WHERE s.batch_id = v_batch_id + WHERE s._load_id = v_batch_id AND t.ticket_no IS NULL; IF v_orphan_ticket_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены segments без соответствующих tickets (batch_id=%): %', + 'DQ FAILED: найдены segments без соответствующих tickets (_load_id=%): %', v_batch_id, v_orphan_ticket_count; END IF; @@ -112,18 +112,18 @@ BEGIN INTO v_orphan_flight_count FROM stg.segments AS s LEFT JOIN stg.flights AS f ON s.flight_id = f.flight_id - WHERE s.batch_id = v_batch_id + WHERE s._load_id = v_batch_id AND f.flight_id IS NULL; IF v_orphan_flight_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены segments без соответствующих flights (batch_id=%): %', + 'DQ FAILED: найдены segments без соответствующих flights (_load_id=%): %', v_batch_id, v_orphan_flight_count; END IF; RAISE NOTICE - 'DQ PASSED: segments ок (batch_id=%): source=% stg=%', + 'DQ PASSED: segments ок (_load_id=%): source=% stg=%', v_batch_id, v_src_count, v_stg_count; diff --git a/sql/stg/segments_load.sql b/sql/stg/segments_load.sql index 86ee77b..1c90dc4 100644 --- a/sql/stg/segments_load.sql +++ b/sql/stg/segments_load.sql @@ -4,19 +4,19 @@ -- CTE для определения максимальной даты загрузки предыдущего батча WITH max_batch_ts AS ( - SELECT COALESCE(MAX(src_created_at_ts), TIMESTAMP '1900-01-01 00:00:00') AS max_ts + SELECT COALESCE(MAX(event_ts), TIMESTAMP '1900-01-01 00:00:00') AS max_ts FROM stg.segments - WHERE batch_id <> '{{ run_id }}'::text - OR batch_id IS NULL + WHERE _load_id <> '{{ run_id }}'::text + OR _load_id IS NULL ) INSERT INTO stg.segments ( ticket_no, flight_id, fare_conditions, price, - src_created_at_ts, - load_dttm, - batch_id + event_ts, + _load_ts, + _load_id ) SELECT ext.ticket_no, @@ -32,11 +32,11 @@ JOIN stg.bookings_ext AS b ON t.book_ref = b.book_ref CROSS JOIN max_batch_ts AS mb WHERE b.book_date > mb.max_ts AND NOT EXISTS ( - -- Идемпотентность: при повторном запуске/ретрае не вставляем повторно те же строки в рамках текущего batch_id. + -- Идемпотентность: при повторном запуске/ретрае не вставляем повторно те же строки в рамках текущего _load_id. -- Считаем ключом строки (ticket_no, flight_id). SELECT 1 FROM stg.segments AS s - WHERE s.batch_id = '{{ run_id }}'::text + WHERE s._load_id = '{{ run_id }}'::text AND s.ticket_no = ext.ticket_no AND s.flight_id = ext.flight_id ); diff --git a/sql/stg/tickets_ddl.sql b/sql/stg/tickets_ddl.sql index 9564ccb..bdf3a14 100644 --- a/sql/stg/tickets_ddl.sql +++ b/sql/stg/tickets_ddl.sql @@ -24,9 +24,9 @@ CREATE TABLE IF NOT EXISTS stg.tickets ( passenger_id TEXT, passenger_name TEXT, outbound TEXT, - src_created_at_ts TIMESTAMP, - load_dttm TIMESTAMP NOT NULL DEFAULT now(), - batch_id TEXT + event_ts TIMESTAMP, + _load_ts TIMESTAMP NOT NULL DEFAULT now(), + _load_id TEXT NOT NULL ) WITH (appendonly=true, orientation=row, compresstype=zstd, compresslevel=1) -- Ключ распределения: book_ref diff --git a/sql/stg/tickets_dq.sql b/sql/stg/tickets_dq.sql index b4b6951..caae980 100644 --- a/sql/stg/tickets_dq.sql +++ b/sql/stg/tickets_dq.sql @@ -11,12 +11,12 @@ DECLARE v_dup_count BIGINT; v_empty_name_count BIGINT; BEGIN - -- Опорная метка: максимум src_created_at_ts среди предыдущих батчей - SELECT max(src_created_at_ts) + -- Опорная метка: максимум event_ts среди предыдущих батчей + SELECT max(event_ts) INTO v_prev_ts FROM stg.tickets - WHERE batch_id <> v_batch_id - OR batch_id IS NULL; + WHERE _load_id <> v_batch_id + OR _load_id IS NULL; -- Количество в источнике (новые билеты в том же окне инкремента, что и загрузка) SELECT COUNT(*) @@ -27,21 +27,21 @@ BEGIN IF v_source_count = 0 THEN -- Пустое окно инкремента допустимо: новых данных может не быть. - -- В этом случае ожидаем, что в текущем batch_id тоже 0 строк. + -- В этом случае ожидаем, что в текущем _load_id тоже 0 строк. SELECT COUNT(*) INTO v_stg_count FROM stg.tickets - WHERE batch_id = v_batch_id; + WHERE _load_id = v_batch_id; IF v_stg_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: источник tickets_ext за окно инкремента пустой, но в stg.tickets есть строки текущего batch_id (batch_id=%): %', + 'DQ FAILED: источник tickets_ext за окно инкремента пустой, но в stg.tickets есть строки текущего _load_id (_load_id=%): %', v_batch_id, v_stg_count; END IF; RAISE NOTICE - 'В источнике tickets_ext нет строк для окна инкремента (book_date > %). Пропускаем DQ проверки (batch_id=%).', + 'В источнике tickets_ext нет строк для окна инкремента (book_date > %). Пропускаем DQ проверки (_load_id=%).', COALESCE(v_prev_ts, TIMESTAMP '1900-01-01 00:00:00'), v_batch_id; RETURN; @@ -51,11 +51,11 @@ BEGIN SELECT COUNT(*) INTO v_stg_count FROM stg.tickets - WHERE batch_id = v_batch_id; + WHERE _load_id = v_batch_id; IF v_source_count <> v_stg_count THEN RAISE EXCEPTION - 'DQ FAILED: несовпадение количества билетов. Источник: %, STG (batch_id=%): %', + 'DQ FAILED: несовпадение количества билетов. Источник: %, STG (_load_id=%): %', v_source_count, v_batch_id, v_stg_count; @@ -67,12 +67,12 @@ BEGIN INTO v_orphan_count FROM stg.tickets AS t LEFT JOIN stg.bookings AS b ON t.book_ref = b.book_ref - WHERE t.batch_id = v_batch_id + WHERE t._load_id = v_batch_id AND b.book_ref IS NULL; IF v_orphan_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены tickets без соответствующих bookings (batch_id=%): %', + 'DQ FAILED: найдены tickets без соответствующих bookings (_load_id=%): %', v_batch_id, v_orphan_count; END IF; @@ -81,12 +81,12 @@ BEGIN SELECT COUNT(*) INTO v_null_count FROM stg.tickets AS t - WHERE t.batch_id = v_batch_id + WHERE t._load_id = v_batch_id AND (t.ticket_no IS NULL OR t.book_ref IS NULL); IF v_null_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены tickets с NULL в обязательных полях (batch_id=%): %', + 'DQ FAILED: найдены tickets с NULL в обязательных полях (_load_id=%): %', v_batch_id, v_null_count; END IF; @@ -95,11 +95,11 @@ BEGIN SELECT COUNT(*) - COUNT(DISTINCT ticket_no) INTO v_dup_count FROM stg.tickets AS t - WHERE t.batch_id = v_batch_id; + WHERE t._load_id = v_batch_id; IF v_dup_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены дубликаты ticket_no (batch_id=%): %', + 'DQ FAILED: найдены дубликаты ticket_no (_load_id=%): %', v_batch_id, v_dup_count; END IF; @@ -108,18 +108,18 @@ BEGIN SELECT COUNT(*) INTO v_empty_name_count FROM stg.tickets AS t - WHERE t.batch_id = v_batch_id + WHERE t._load_id = v_batch_id AND (t.passenger_name IS NULL OR t.passenger_name = ''); IF v_empty_name_count <> 0 THEN RAISE EXCEPTION - 'DQ FAILED: найдены tickets с пустым именем пассажира (batch_id=%): %', + 'DQ FAILED: найдены tickets с пустым именем пассажира (_load_id=%): %', v_batch_id, v_empty_name_count; END IF; RAISE NOTICE - 'DQ PASSED: tickets ок (batch_id=%): source=% stg=%', + 'DQ PASSED: tickets ок (_load_id=%): source=% stg=%', v_batch_id, v_source_count, v_stg_count; diff --git a/sql/stg/tickets_load.sql b/sql/stg/tickets_load.sql index 925ac80..551aee0 100644 --- a/sql/stg/tickets_load.sql +++ b/sql/stg/tickets_load.sql @@ -3,10 +3,10 @@ -- CTE для определения максимальной даты загрузки предыдущего батча WITH max_batch_ts AS ( - SELECT COALESCE(MAX(src_created_at_ts), TIMESTAMP '1900-01-01 00:00:00') AS max_ts + SELECT COALESCE(MAX(event_ts), TIMESTAMP '1900-01-01 00:00:00') AS max_ts FROM stg.tickets - WHERE batch_id <> '{{ run_id }}'::text - OR batch_id IS NULL + WHERE _load_id <> '{{ run_id }}'::text + OR _load_id IS NULL ) INSERT INTO stg.tickets ( ticket_no, @@ -14,9 +14,9 @@ INSERT INTO stg.tickets ( passenger_id, passenger_name, outbound, - src_created_at_ts, - load_dttm, - batch_id + event_ts, + _load_ts, + _load_id ) SELECT ext.ticket_no, diff --git a/tests/test_dags_smoke.py b/tests/test_dags_smoke.py index 6ade3ef..8d645e5 100644 --- a/tests/test_dags_smoke.py +++ b/tests/test_dags_smoke.py @@ -119,14 +119,14 @@ def test_bookings_to_gp_stage_dag_structure(): _assert_direct_edge(dag, load_task_id, dq_task_id) # Барьеры по данным (не обязательно прямые рёбра). - # routes_dq использует airports/airplanes текущего batch_id. + # routes_dq использует airports/airplanes текущего _load_id. _assert_reachable(dag, "check_airports_dq", "check_routes_dq") _assert_reachable(dag, "check_airplanes_dq", "check_routes_dq") - # seats_dq использует airplanes текущего batch_id. + # seats_dq использует airplanes текущего _load_id. _assert_reachable(dag, "check_airplanes_dq", "check_seats_dq") - # flights_dq использует routes текущего batch_id. + # flights_dq использует routes текущего _load_id. _assert_reachable(dag, "check_routes_dq", "check_flights_dq") # segments_dq проверяет наличие flights (STG-история); для первой загрузки flights должны быть до segments. diff --git a/tests/test_ods_snapshot_integration.py b/tests/test_ods_snapshot_integration.py index f0ca294..2fb49e2 100644 --- a/tests/test_ods_snapshot_integration.py +++ b/tests/test_ods_snapshot_integration.py @@ -67,15 +67,15 @@ def test_snapshot_airports_contract_upsert_delete_and_dq() -> None: DROP TABLE IF EXISTS {ods_table}; CREATE TABLE {stg_table} ( - airport_code TEXT, - airport_name TEXT, - city TEXT, - country TEXT, - coordinates TEXT, - timezone TEXT, - src_created_at_ts TIMESTAMP, - load_dttm TIMESTAMP, - batch_id TEXT + airport_code TEXT, + airport_name TEXT, + city TEXT, + country TEXT, + coordinates TEXT, + timezone TEXT, + event_ts TIMESTAMP, + _load_ts TIMESTAMP, + _load_id TEXT ); CREATE TABLE {ods_table} ( @@ -96,7 +96,7 @@ def test_snapshot_airports_contract_upsert_delete_and_dq() -> None: f""" INSERT INTO {stg_table} ( airport_code, airport_name, city, country, coordinates, timezone, - src_created_at_ts, load_dttm, batch_id + event_ts, _load_ts, _load_id ) VALUES ('AAA', '{{"en": "Airport A", "ru": "Аэропорт A"}}', '{{"en": "City A", "ru": "Город A"}}', '{{"en": "Country A", "ru": "Страна A"}}', '(0,0)', 'UTC', now(), now(), 'batch_1'), ('BBB', '{{"en": "Airport B", "ru": "Аэропорт B"}}', '{{"en": "City B", "ru": "Город B"}}', '{{"en": "Country B", "ru": "Страна B"}}', '(1,1)', 'UTC', now(), now(), 'batch_1'); @@ -118,7 +118,7 @@ def test_snapshot_airports_contract_upsert_delete_and_dq() -> None: f""" INSERT INTO {stg_table} ( airport_code, airport_name, city, country, coordinates, timezone, - src_created_at_ts, load_dttm, batch_id + event_ts, _load_ts, _load_id ) VALUES ('AAA', '{{"en": "Airport A v2", "ru": "Аэропорт A v2"}}', '{{"en": "City A", "ru": "Город A"}}', '{{"en": "Country A", "ru": "Страна A"}}', '(0,0)', 'UTC', now(), now(), 'batch_2'), ('CCC', '{{"en": "Airport C", "ru": "Аэропорт C"}}', '{{"en": "City C", "ru": "Город C"}}', '{{"en": "Country C", "ru": "Страна C"}}', '(2,2)', 'UTC', now(), now(), 'batch_2'); diff --git a/tests/test_ods_sql_contract.py b/tests/test_ods_sql_contract.py index 32a03b5..41726ff 100644 --- a/tests/test_ods_sql_contract.py +++ b/tests/test_ods_sql_contract.py @@ -28,7 +28,7 @@ def test_snapshot_dq_checks_extra_keys() -> None: def test_ods_batch_resolver_uses_consistent_snapshot_batches() -> None: - """Резолвер батча должен искать batch_id, общий для всех snapshot-таблиц STG.""" + """Резолвер батча должен искать _load_id, общий для всех snapshot-таблиц STG.""" dag_code = _read("airflow/dags/bookings_to_gp_ods.py") for table_name in ("stg.airports", "stg.airplanes", "stg.routes", "stg.seats"):