Отладка потоков

airflow/dags/bookings_stg_ddl.py

@@ -82,19 +82,15 @@ with DAG(
         sql="stg/boarding_passes_ddl.sql",
     )
 
-    # Сначала создаются справочники, затем транзакционные таблицы
+    # Сначала создаются справочники, затем транзакционные таблицы (последовательно)
     (
         apply_stg_bookings_ddl
         >> apply_stg_tickets_ddl
-        >> [
-            apply_stg_airports_ddl,
-            apply_stg_airplanes_ddl,
-            apply_stg_routes_ddl,
-            apply_stg_seats_ddl,
-        ]
-        >> [
-            apply_stg_flights_ddl,
-            apply_stg_segments_ddl,
-            apply_stg_boarding_passes_ddl,
-        ]
+        >> apply_stg_airports_ddl
+        >> apply_stg_airplanes_ddl
+        >> apply_stg_routes_ddl
+        >> apply_stg_seats_ddl
+        >> apply_stg_flights_ddl
+        >> apply_stg_segments_ddl
+        >> apply_stg_boarding_passes_ddl
     )

airflow/dags/bookings_to_gp_stage.py

@@ -190,20 +190,16 @@ with DAG(
     generate_bookings_day >> load_bookings_to_stg >> check_row_counts
     check_row_counts >> load_tickets_to_stg >> check_tickets_dq
 
-    # Затем загружаются справочники
-    check_tickets_dq >> [
-        (load_airports_to_stg >> check_airports_dq),
-        (load_airplanes_to_stg >> check_airplanes_dq),
-        (load_routes_to_stg >> check_routes_dq),
-        (load_seats_to_stg >> check_seats_dq),
-    ]
+    # Затем загружаются справочники (последовательная загрузка)
+    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_airports_dq, check_airplanes_dq, check_routes_dq, check_seats_dq] >> [
-        (load_flights_to_stg >> check_flights_dq),
-        (load_segments_to_stg >> check_segments_dq),
-        (load_boarding_passes_to_stg >> check_boarding_passes_dq),
-    ]
+    # Затем загружаются транзакции (последовательная загрузка)
+    check_seats_dq >> load_flights_to_stg >> check_flights_dq
+    check_flights_dq >> load_segments_to_stg >> check_segments_dq
+    check_segments_dq >> load_boarding_passes_to_stg >> check_boarding_passes_dq
 
     # В конце финальный лог
-    [check_flights_dq, check_segments_dq, check_boarding_passes_dq] >> finish_summary
+    check_boarding_passes_dq >> finish_summary