71 lines
2.4 KiB
Python
71 lines
2.4 KiB
Python
"""Готовый DAG первой лабораторной: общий снимок справочников без генерации дней."""
|
|
|
|
from datetime import timedelta
|
|
|
|
import pendulum
|
|
from airflow.providers.postgres.operators.postgres import PostgresOperator
|
|
|
|
from airflow import DAG
|
|
|
|
with DAG(
|
|
dag_id="lab_pxf_airports",
|
|
start_date=pendulum.datetime(2017, 1, 1, tz="UTC"),
|
|
schedule=None,
|
|
catchup=False,
|
|
max_active_runs=1,
|
|
template_searchpath="/sql",
|
|
default_args={
|
|
"owner": "airflow",
|
|
"retries": 1,
|
|
"retry_delay": timedelta(seconds=30),
|
|
},
|
|
tags=["lab", "bookings", "pxf"],
|
|
description="Лабораторная PXF: четыре снимка STG с общим Run ID для ODS",
|
|
) as dag:
|
|
load_airports_to_stg = PostgresOperator(
|
|
task_id="load_airports_to_stg",
|
|
postgres_conn_id="greenplum_conn",
|
|
sql="stg/airports_load.sql",
|
|
)
|
|
check_airports_dq = PostgresOperator(
|
|
task_id="check_airports_dq",
|
|
postgres_conn_id="greenplum_conn",
|
|
sql="stg/airports_dq.sql",
|
|
)
|
|
load_airplanes_to_stg = PostgresOperator(
|
|
task_id="load_airplanes_to_stg",
|
|
postgres_conn_id="greenplum_conn",
|
|
sql="stg/airplanes_load.sql",
|
|
)
|
|
check_airplanes_dq = PostgresOperator(
|
|
task_id="check_airplanes_dq",
|
|
postgres_conn_id="greenplum_conn",
|
|
sql="stg/airplanes_dq.sql",
|
|
)
|
|
load_routes_to_stg = PostgresOperator(
|
|
task_id="load_routes_to_stg",
|
|
postgres_conn_id="greenplum_conn",
|
|
sql="stg/routes_load.sql",
|
|
)
|
|
check_routes_dq = PostgresOperator(
|
|
task_id="check_routes_dq",
|
|
postgres_conn_id="greenplum_conn",
|
|
sql="stg/routes_dq.sql",
|
|
)
|
|
load_seats_to_stg = PostgresOperator(
|
|
task_id="load_seats_to_stg",
|
|
postgres_conn_id="greenplum_conn",
|
|
sql="stg/seats_load.sql",
|
|
)
|
|
check_seats_dq = PostgresOperator(
|
|
task_id="check_seats_dq",
|
|
postgres_conn_id="greenplum_conn",
|
|
sql="stg/seats_dq.sql",
|
|
)
|
|
|
|
load_airports_to_stg >> check_airports_dq
|
|
load_airplanes_to_stg >> check_airplanes_dq
|
|
# Готовим один общий батч: DQ маршрутов и мест читает связанные справочники.
|
|
[check_airports_dq, check_airplanes_dq] >> load_routes_to_stg >> check_routes_dq
|
|
check_airplanes_dq >> load_seats_to_stg >> check_seats_dq
|