From 8deb174221110318b215b498b37a024bf2ef4d95 Mon Sep 17 00:00:00 2001 From: syahril Date: Wed, 21 Oct 2026 14:55:57 +0700 Subject: [PATCH] fix(dag): add sys.path relative import and correct duckdb partition column --- configs/e2e_modernized_ingestion_flow.yaml | 12 ++++++------ dags/dag_factory.py | 6 ++++++ 2 files changed, 12 insertions(+), 6 deletions(-) diff --git a/configs/e2e_modernized_ingestion_flow.yaml b/configs/e2e_modernized_ingestion_flow.yaml index 8a9af37..8004c59 100644 --- a/configs/e2e_modernized_ingestion_flow.yaml +++ b/configs/e2e_modernized_ingestion_flow.yaml @@ -71,15 +71,15 @@ tasks: target_table: bronze_environmental_permits table_name: bronze_environmental_permits primary_keys: permit_id - query: SELECT id AS permit_id, permit_number, company_name, permit_type, status, + query: SELECT permit_id, permit_number, company_name, activity_type AS permit_type, status, issued_date, updated_at FROM read_parquet('s3://dlh-bronze/staging/environmental_permits/*.parquet') - QUALIFY ROW_NUMBER() OVER (PARTITION BY id ORDER BY updated_at DESC) = 1 - sql_query: SELECT id AS permit_id, permit_number, company_name, permit_type, status, + QUALIFY ROW_NUMBER() OVER (PARTITION BY permit_id ORDER BY updated_at DESC) = 1 + sql_query: SELECT permit_id, permit_number, company_name, activity_type AS permit_type, status, issued_date, updated_at FROM read_parquet('s3://dlh-bronze/staging/environmental_permits/*.parquet') - QUALIFY ROW_NUMBER() OVER (PARTITION BY id ORDER BY updated_at DESC) = 1 - transformation_sql: SELECT id AS permit_id, permit_number, company_name, permit_type, + QUALIFY ROW_NUMBER() OVER (PARTITION BY permit_id ORDER BY updated_at DESC) = 1 + transformation_sql: SELECT permit_id, permit_number, company_name, activity_type AS permit_type, status, issued_date, updated_at FROM read_parquet('s3://dlh-bronze/staging/environmental_permits/*.parquet') - QUALIFY ROW_NUMBER() OVER (PARTITION BY id ORDER BY updated_at DESC) = 1 + QUALIFY ROW_NUMBER() OVER (PARTITION BY permit_id ORDER BY updated_at DESC) = 1 - task_id: node-6733 type: dbt_transform label: dbt Silver Cleansed Permits Enriched diff --git a/dags/dag_factory.py b/dags/dag_factory.py index 106ad72..7991feb 100644 --- a/dags/dag_factory.py +++ b/dags/dag_factory.py @@ -1,9 +1,15 @@ import os +import sys import glob import yaml from datetime import datetime, timedelta from airflow import DAG +# Ensure bundle/dags directory is on sys.path for relative helper imports +current_dir = os.path.dirname(os.path.abspath(__file__)) +if current_dir not in sys.path: + sys.path.insert(0, current_dir) + try: from airflow.providers.standard.operators.bash import BashOperator from airflow.providers.standard.operators.python import PythonOperator