ddmitry/airflow-greenplum
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(all): унифицированы служебные поля STG — переход на канонический нейминг

Browse Source 
- Зачем:
  - STG-слой использовал legacy-имена (batch_id, load_dttm, src_created_at_ts),
    тогда как ODS/DDS/DM уже работали с каноном (_load_id, _load_ts, event_ts).
    Студент видел разные имена для одного понятия — это убрано.
- Что:
  - переименованы колонки в 9 STG DDL: batch_id→_load_id, load_dttm→_load_ts,
    src_created_at_ts→event_ts; добавлен NOT NULL для _load_id во всех таблицах.
  - обновлены 9 STG Load, 9 STG DQ, 9 ODS Load, 9 ODS DQ (INSERT/SELECT/WHERE).
  - обновлены DAG-файлы bookings_to_gp_stage.py и bookings_to_gp_ods.py
    (встроенный SQL резолвера, комментарии; Python-идентификаторы не тронуты).
  - обновлены тесты и ~15 документов (naming_conventions, PRD, db_schema,
    design-docs, qa-plan, README, TESTING и др.).
- Проверка:
  - grep -rn 'load_dttm\|src_created_at_ts' sql/ airflow/ tests/ — 0 совпадений.
  - make test — 4 passed.
  - e2e-etl: day1 прошёл полностью, day2 стартовал без ошибок.
This commit is contained in:
ddadmin
2026-03-10 00:07:08 +03:00
parent 3de4db8aa6
commit 5972bcc2d8
64 changed files with 582 additions and 484 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+1 -1
View File
@@ -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;
TESTING.md
+1 -1
View File
@@ -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» режиме;
airflow/dags/bookings_to_gp_ods.py
+19 -19
View File
@@ -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
airflow/dags/bookings_to_gp_stage.py
+1 -1
View File
@@ -12,7 +12,7 @@ from __future__ import annotations
Каждый запуск DAG работает как «шаг по времени вперёд»:
- генератор в демо-БД bookings добавляет следующий учебный день после max(book_date);
- загрузка в Greenplum берёт все строки, появившиеся после предыдущих батчей;
- `run_id` используется как метка запуска (в `batch_id`, в логах и DQ).
- `run_id` используется как метка запуска (в `_load_id`, в логах и DQ).
Важно: для инкрементальных таблиц «пустое окно инкремента» допустимо (это не ошибка).
docs/bookings_to_gp_ods.md
+1 -1
View File
@@ -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`).
docs/bookings_to_gp_stage.md
+9 -9
View File
@@ -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;
```
docs/internal/PRD.md
+2 -2
View File
@@ -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)
---
docs/internal/architecture_review.md
+8 -9
View File
@@ -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-файла |
docs/internal/bookings_ods_design.md
+30 -30
View File
@@ -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 = '<stg_batch_id>'
WHERE _load_id = '<stg_batch_id>'
) s
WHERE NOT EXISTS (
SELECT 1
docs/internal/bookings_stg_code_review.md
+8 -8
View File
@@ -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-тест реального графа (минимальный полезный уровень)
docs/internal/bookings_stg_design.md
+12 -12
View File
@@ -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`
docs/internal/db_schema.md
+16 -16
View File
@@ -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 обязательных полей, ссылочная целостность
docs/internal/naming_conventions.md
+4 -18
View File
@@ -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` в одном смысле.
docs/internal/qa-plan.md
+19 -24
View File
@@ -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
docs/internal/stg_naming_unification_plan.md
+118
View File
@@ -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 механические замены в каждом
educational-tasks.md
+2 -2
View File
@@ -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 слоёв будут добавлены позже.
sql/ods/airplanes_dq.sql
+7 -7
View File
@@ -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 $$;
sql/ods/airplanes_load.sql
+2 -2
View File
@@ -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,
sql/ods/airports_dq.sql
+7 -7
View File
@@ -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 $$;
sql/ods/airports_load.sql
+2 -2
View File
@@ -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,
sql/ods/boarding_passes_dq.sql
+4 -4
View File
@@ -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 $$;
sql/ods/boarding_passes_load.sql
+9 -9
View File
@@ -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
sql/ods/bookings_dq.sql
+4 -4
View File
@@ -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 $$;
sql/ods/bookings_load.sql
+9 -9
View File
@@ -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
sql/ods/flights_dq.sql
+4 -4
View File
@@ -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 $$;
sql/ods/flights_load.sql
+16 -16
View File
@@ -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
sql/ods/routes_dq.sql
+7 -7
View File
@@ -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 $$;
sql/ods/routes_load.sql
+2 -2
View File
@@ -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,
sql/ods/seats_dq.sql
+7 -7
View File
@@ -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 $$;
sql/ods/seats_load.sql
+2 -2
View File
@@ -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,
sql/ods/segments_dq.sql
+4 -4
View File
@@ -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 $$;
sql/ods/segments_load.sql
+9 -9
View File
@@ -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
sql/ods/tickets_dq.sql
+4 -4
View File
@@ -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 $$;
sql/ods/tickets_load.sql
+9 -9
View File
@@ -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
sql/stg/airplanes_ddl.sql
+3 -3
View File
@@ -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
sql/stg/airplanes_dq.sql
+6 -6
View File
@@ -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;
sql/stg/airplanes_load.sql
+6 -6
View File
@@ -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
);
sql/stg/airports_ddl.sql
+3 -3
View File
@@ -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
sql/stg/airports_dq.sql
+6 -6
View File
@@ -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;
sql/stg/airports_load.sql
+6 -6
View File
@@ -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
);
sql/stg/boarding_passes_ddl.sql
+3 -3
View File
@@ -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
sql/stg/boarding_passes_dq.sql
+12 -12
View File
@@ -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 $$;
sql/stg/boarding_passes_load.sql
+6 -6
View File
@@ -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
);
sql/stg/bookings_ddl.sql
+3 -3
View File
@@ -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
sql/stg/bookings_dq.sql
+15 -15
View File
@@ -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;
sql/stg/bookings_load.sql
+10 -10
View File
@@ -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
);
sql/stg/flights_ddl.sql
+3 -3
View File
@@ -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
sql/stg/flights_dq.sql
+17 -17
View File
@@ -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;
sql/stg/flights_load.sql
+10 -10
View File
@@ -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
);
sql/stg/routes_ddl.sql
+3 -3
View File
@@ -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
sql/stg/routes_dq.sql
+15 -15
View File
@@ -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;
sql/stg/routes_load.sql
+6 -6
View File
@@ -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
);
sql/stg/seats_ddl.sql
+3 -3
View File
@@ -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
sql/stg/seats_dq.sql
+9 -9
View File
@@ -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;
sql/stg/seats_load.sql
+6 -6
View File
@@ -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
);
sql/stg/segments_ddl.sql
+3 -3
View File
@@ -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
sql/stg/segments_dq.sql
+18 -18
View File
@@ -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;
sql/stg/segments_load.sql
+8 -8
View File
@@ -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
);
sql/stg/tickets_ddl.sql
+3 -3
View File
@@ -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
sql/stg/tickets_dq.sql
+19 -19
View File
@@ -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;
sql/stg/tickets_load.sql
+6 -6
View File
@@ -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,
tests/test_dags_smoke.py
+3 -3
View File
@@ -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.
tests/test_ods_snapshot_integration.py
+5 -5
View File
@@ -73,9 +73,9 @@ def test_snapshot_airports_contract_upsert_delete_and_dq() -> None:
country TEXT,
coordinates TEXT,
timezone TEXT,
src_created_at_ts TIMESTAMP,
load_dttm TIMESTAMP,
batch_id 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');
tests/test_ods_sql_contract.py
+1 -1
View File
@@ -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"):