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

фикс вфп инициализации ddl в gp

Browse Source
This commit is contained in:
ddadmin
2025-12-10 22:31:12 +03:00
parent 6e2a43a900
commit fbf9b118e6
4 changed files with 9 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+1 -1
View File
@@ -285,7 +285,7 @@ docker compose -f docker-compose.yml exec bookings-db bash -lc 'PGPASSWORD="$POS
load_bookings_to_stg = PostgresOperator( load_bookings_to_stg = PostgresOperator(
task_id="load_bookings_to_stg", task_id="load_bookings_to_stg",
postgres_conn_id="greenplum_conn", postgres_conn_id="greenplum_conn",
sql="/sql/stg/bookings_load.sql", sql="stg/bookings_load.sql",
params={"batch_id": "{{ ds_nodash }}"}, params={"batch_id": "{{ ds_nodash }}"},
) )
``` ```
airflow/dags/bookings_stg_ddl.py
+2 -1
View File
@@ -20,6 +20,7 @@ with DAG(
start_date=datetime(2024, 1, 1), start_date=datetime(2024, 1, 1),
schedule=None, schedule=None,
catchup=False, catchup=False,
template_searchpath="/sql",
default_args=default_args, default_args=default_args,
tags=["demo", "greenplum", "ddl", "bookings", "stg"], tags=["demo", "greenplum", "ddl", "bookings", "stg"],
description="Создаёт/обновляет stg.bookings_ext и stg.bookings для учебного DAG", description="Создаёт/обновляет stg.bookings_ext и stg.bookings для учебного DAG",
@@ -27,5 +28,5 @@ with DAG(
apply_stg_bookings_ddl = PostgresOperator( apply_stg_bookings_ddl = PostgresOperator(
task_id="apply_stg_bookings_ddl", task_id="apply_stg_bookings_ddl",
postgres_conn_id=GREENPLUM_CONN_ID, postgres_conn_id=GREENPLUM_CONN_ID,
sql="/sql/stg/bookings_ddl.sql", sql="stg/bookings_ddl.sql",
) )
airflow/dags/bookings_to_gp_stage.py
+4 -3
View File
@@ -51,6 +51,7 @@ with DAG(
start_date=datetime(2024, 1, 1), start_date=datetime(2024, 1, 1),
schedule=None, schedule=None,
catchup=False, catchup=False,
template_searchpath="/sql",
default_args=default_args, default_args=default_args,
tags=["demo", "bookings", "greenplum", "stg"], tags=["demo", "bookings", "greenplum", "stg"],
description="Учебный DAG: загрузка из bookings-db в stg.bookings (Greenplum)", description="Учебный DAG: загрузка из bookings-db в stg.bookings (Greenplum)",
@@ -59,14 +60,14 @@ with DAG(
generate_bookings_day = PostgresOperator( generate_bookings_day = PostgresOperator(
task_id="generate_bookings_day", task_id="generate_bookings_day",
postgres_conn_id=BOOKINGS_CONN_ID, postgres_conn_id=BOOKINGS_CONN_ID,
sql="/sql/src/bookings_generate_day_if_missing.sql", sql="src/bookings_generate_day_if_missing.sql",
) )
# 2. Загружаем инкремент из stg.bookings_ext в stg.bookings # 2. Загружаем инкремент из stg.bookings_ext в stg.bookings
load_bookings_to_stg = PostgresOperator( load_bookings_to_stg = PostgresOperator(
task_id="load_bookings_to_stg", task_id="load_bookings_to_stg",
postgres_conn_id=GREENPLUM_CONN_ID, postgres_conn_id=GREENPLUM_CONN_ID,
sql="/sql/stg/bookings_load.sql", sql="stg/bookings_load.sql",
params={ params={
"batch_id": "{{ ds_nodash }}", "batch_id": "{{ ds_nodash }}",
}, },
@@ -76,7 +77,7 @@ with DAG(
check_row_counts = PostgresOperator( check_row_counts = PostgresOperator(
task_id="check_row_counts", task_id="check_row_counts",
postgres_conn_id=GREENPLUM_CONN_ID, postgres_conn_id=GREENPLUM_CONN_ID,
sql="/sql/stg/bookings_dq.sql", sql="stg/bookings_dq.sql",
params={ params={
"batch_id": "{{ ds_nodash }}", "batch_id": "{{ ds_nodash }}",
}, },
airflow/dags/ddl_greenplum_base.py
+2 -1
View File
@@ -20,6 +20,7 @@ with DAG(
start_date=datetime(2024, 1, 1), start_date=datetime(2024, 1, 1),
schedule=None, schedule=None,
catchup=False, catchup=False,
template_searchpath="/sql",
default_args=default_args, default_args=default_args,
tags=["demo", "greenplum", "ddl", "orders"], tags=["demo", "greenplum", "ddl", "orders"],
description="Создаёт/обновляет базовую таблицу orders в схеме public", description="Создаёт/обновляет базовую таблицу orders в схеме public",
@@ -27,5 +28,5 @@ with DAG(
apply_orders_ddl = PostgresOperator( apply_orders_ddl = PostgresOperator(
task_id="apply_orders_ddl", task_id="apply_orders_ddl",
postgres_conn_id=GREENPLUM_CONN_ID, postgres_conn_id=GREENPLUM_CONN_ID,
sql="/sql/base/orders_ddl.sql", sql="base/orders_ddl.sql",
) )