refactor(dags): параллелизован граф DAG bookings_to_gp_stage
- Зачем:
- линейный граф маскировал реальные зависимости данных; для эталонного
стенда важно показать менти параллельный граф там, где данные независимы.
- Что:
- airports и airplanes грузятся параллельно после check_tickets_dq.
- routes ждёт обоих (DQ проверяет ссылочную целостность на оба справочника).
- seats зависит только от airplanes и работает параллельно с веткой
routes → flights → segments → boarding_passes.
- finish_summary ждёт обе ветки (check_boarding_passes_dq + check_seats_dq).
- datetime.utcnow() заменён на datetime.now(UTC) в csv_to_greenplum.py.
- smoke-тесты дополнены проверкой параллельности и второй ветки.
- документация обновлена с ASCII-схемой нового графа.
- Проверка:
- make test (11 passed, 4 skipped), make lint — чисто.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -15,6 +15,23 @@ from __future__ import annotations
|
||||
- `run_id` используется как метка запуска (в `batch_id`, в логах и DQ).
|
||||
|
||||
Важно: для инкрементальных таблиц «пустое окно инкремента» допустимо (это не ошибка).
|
||||
|
||||
Граф зависимостей (параллельный там, где данные независимы):
|
||||
|
||||
generate_bookings_day → load_bookings → check_bookings_dq
|
||||
→ load_tickets → check_tickets_dq
|
||||
├─ load_airports → check_airports_dq ─┐
|
||||
│ ├─ load_routes → check_routes_dq
|
||||
├─ load_airplanes → check_airplanes_dq ┤ → load_flights → check_flights_dq
|
||||
│ │ → load_segments → check_segments_dq
|
||||
│ │ → load_boarding_passes → check_bp_dq ─┐
|
||||
│ └─ load_seats → check_seats_dq ──────────────────────┤
|
||||
│ ▼
|
||||
└──────────────────────────────────────────────────────────────────────────── finish_summary
|
||||
|
||||
airports и airplanes грузятся параллельно (они не зависят друг от друга).
|
||||
routes зависит от обоих (DQ проверяет ссылочную целостность на airports и airplanes).
|
||||
seats зависит только от airplanes (DQ проверяет airplane_code → airplanes).
|
||||
"""
|
||||
|
||||
from datetime import timedelta
|
||||
@@ -188,22 +205,30 @@ with DAG(
|
||||
python_callable=_finish_summary,
|
||||
)
|
||||
|
||||
# Сначала загружаются и проверяются bookings и tickets
|
||||
# === Этап 1. Транзакции: bookings → tickets (последовательно, т.к. tickets зависят от bookings) ===
|
||||
generate_bookings_day >> load_bookings_to_stg >> check_row_counts
|
||||
check_row_counts >> load_tickets_to_stg >> check_tickets_dq
|
||||
|
||||
# Затем загружаются справочники.
|
||||
# Для простоты (и более понятных логов для новичков) делаем это последовательно.
|
||||
# Если позже понадобится ускорить DAG, эти шаги можно распараллелить, сохранив зависимости.
|
||||
# === Этап 2. Справочники (параллельно, где данные независимы) ===
|
||||
# airports и airplanes не зависят друг от друга — грузим параллельно.
|
||||
check_tickets_dq >> load_airports_to_stg >> check_airports_dq
|
||||
check_airports_dq >> load_airplanes_to_stg >> check_airplanes_dq
|
||||
check_airplanes_dq >> load_routes_to_stg >> check_routes_dq
|
||||
check_routes_dq >> load_seats_to_stg >> check_seats_dq
|
||||
check_tickets_dq >> load_airplanes_to_stg >> check_airplanes_dq
|
||||
|
||||
# Затем загружаются транзакции (тоже последовательно, по тем же причинам).
|
||||
check_seats_dq >> load_flights_to_stg >> check_flights_dq
|
||||
# routes зависит от airports И airplanes (DQ проверяет ссылочную целостность).
|
||||
[check_airports_dq, check_airplanes_dq] >> load_routes_to_stg >> check_routes_dq
|
||||
|
||||
# seats зависит только от airplanes (DQ проверяет ссылочную целостность).
|
||||
check_airplanes_dq >> load_seats_to_stg >> check_seats_dq
|
||||
|
||||
# === Этап 3. Транзакции (последовательно, каждая зависит от предыдущей) ===
|
||||
# flights зависят от routes (DQ проверяет ссылочную целостность route_no → routes).
|
||||
check_routes_dq >> load_flights_to_stg >> check_flights_dq
|
||||
|
||||
# segments зависят от flights и tickets (DQ проверяет обе ссылки).
|
||||
check_flights_dq >> load_segments_to_stg >> check_segments_dq
|
||||
|
||||
# boarding_passes зависят от segments и tickets (DQ проверяет обе ссылки).
|
||||
check_segments_dq >> load_boarding_passes_to_stg >> check_boarding_passes_dq
|
||||
|
||||
# В конце финальный лог
|
||||
check_boarding_passes_dq >> finish_summary
|
||||
# === Финал: ждём завершения ВСЕХ веток ===
|
||||
[check_boarding_passes_dq, check_seats_dq] >> finish_summary
|
||||
|
||||
@@ -3,7 +3,7 @@ from __future__ import annotations
|
||||
import logging
|
||||
import os
|
||||
import random
|
||||
from datetime import datetime, timedelta
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from pathlib import Path
|
||||
from typing import List
|
||||
|
||||
@@ -37,11 +37,11 @@ def _create_table() -> None:
|
||||
def _generate_csv(rows: int, csv_dir: Path) -> str:
|
||||
"""Генерирует CSV c заказами с помощью pandas и сохраняет на диск."""
|
||||
csv_dir.mkdir(parents=True, exist_ok=True)
|
||||
timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
|
||||
timestamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S")
|
||||
csv_path = csv_dir / f"orders_{timestamp}.csv"
|
||||
|
||||
# Генерируем данные в pandas-стиле
|
||||
base_order_id = int(datetime.utcnow().timestamp() * 1_000)
|
||||
base_order_id = int(datetime.now(UTC).timestamp() * 1_000)
|
||||
|
||||
# Создаём DataFrame с использованием pandas методов
|
||||
df = pd.DataFrame(
|
||||
@@ -52,7 +52,7 @@ def _generate_csv(rows: int, csv_dir: Path) -> str:
|
||||
),
|
||||
# Временные метки с интервалом в 1 секунду в обратном порядке
|
||||
"order_ts": pd.date_range(
|
||||
end=datetime.utcnow(), periods=rows, freq="1S"
|
||||
end=datetime.now(UTC), periods=rows, freq="1S"
|
||||
).sort_values(ascending=False),
|
||||
# Случайные customer_id от 1 до 1000
|
||||
"customer_id": pd.Series(
|
||||
|
||||
Reference in New Issue
Block a user