From fa1065c670013ee418b8db6273bce80e6e2ef961 Mon Sep 17 00:00:00 2001 From: syahril Date: Mon, 19 Oct 2026 02:07:39 +0700 Subject: [PATCH] fix(pipeline): enable dynamic table extraction, dbt bronze models, and valid quality gate assertions --- configs/e2e_orders_master_medallion_flow.yaml | 28 ++++--------- dags/helpers/dbt_trino_operator.py | 4 +- dags/helpers/sling_helper.py | 39 +++++++++++++++---- dbt/dbt_project.yml | 3 ++ dbt/models/bronze/bronze_ref_facilities.sql | 15 +++++++ dbt/models/bronze/bronze_waste_orders.sql | 16 ++++++++ dbt/models/sources.yml | 8 ++++ 7 files changed, 84 insertions(+), 29 deletions(-) create mode 100644 dbt/models/bronze/bronze_ref_facilities.sql create mode 100644 dbt/models/bronze/bronze_waste_orders.sql diff --git a/configs/e2e_orders_master_medallion_flow.yaml b/configs/e2e_orders_master_medallion_flow.yaml index e6595ca..655517a 100644 --- a/configs/e2e_orders_master_medallion_flow.yaml +++ b/configs/e2e_orders_master_medallion_flow.yaml @@ -67,11 +67,8 @@ tasks: table_max_rows: 1000000 ge_col_table: stasiun_spku ge_target_column: stasiun_id - ge_column_test_type: not_null - sql_assertion: SELECT count(*) FROM silver.table WHERE status NOT IN ('APPROVED', - 'IN_REVIEW', 'REJECTED') - sql: SELECT count(*) FROM silver.table WHERE status NOT IN ('APPROVED', 'IN_REVIEW', - 'REJECTED') + sql_assertion: SELECT count(*) FROM public.raw_waste_orders WHERE order_status NOT IN ('COMPLETED', 'PENDING', 'CANCELLED') + sql: SELECT count(*) FROM public.raw_waste_orders WHERE order_status NOT IN ('COMPLETED', 'PENDING', 'CANCELLED') expected: 0 error_message: Unexpected rows detected in table suite: bronze @@ -118,11 +115,8 @@ tasks: table_max_rows: 1000000 ge_col_table: stasiun_spku ge_target_column: stasiun_id - ge_column_test_type: not_null - sql_assertion: SELECT count(*) FROM silver.table WHERE status NOT IN ('APPROVED', - 'IN_REVIEW', 'REJECTED') - sql: SELECT count(*) FROM silver.table WHERE status NOT IN ('APPROVED', 'IN_REVIEW', - 'REJECTED') + sql_assertion: SELECT count(*) FROM public.raw_waste_orders WHERE volume_ton <= 0 + sql: SELECT count(*) FROM public.raw_waste_orders WHERE volume_ton <= 0 expected: 0 error_message: Unexpected rows detected in table suite: bronze @@ -178,11 +172,8 @@ tasks: table_max_rows: 1000000 ge_col_table: stasiun_spku ge_target_column: stasiun_id - ge_column_test_type: not_null - sql_assertion: SELECT count(*) FROM silver.table WHERE status NOT IN ('APPROVED', - 'IN_REVIEW', 'REJECTED') - sql: SELECT count(*) FROM silver.table WHERE status NOT IN ('APPROVED', 'IN_REVIEW', - 'REJECTED') + sql_assertion: SELECT count(*) FROM public.raw_waste_orders WHERE facility_id IS NULL + sql: SELECT count(*) FROM public.raw_waste_orders WHERE facility_id IS NULL expected: 0 error_message: Unexpected rows detected in table suite: bronze @@ -235,11 +226,8 @@ tasks: table_max_rows: 1000000 ge_col_table: stasiun_spku ge_target_column: stasiun_id - ge_column_test_type: not_null - sql_assertion: SELECT count(*) FROM silver.table WHERE status NOT IN ('APPROVED', - 'IN_REVIEW', 'REJECTED') - sql: SELECT count(*) FROM silver.table WHERE status NOT IN ('APPROVED', 'IN_REVIEW', - 'REJECTED') + sql_assertion: SELECT count(*) FROM public.ref_waste_facilities WHERE max_daily_capacity_ton <= 0 + sql: SELECT count(*) FROM public.ref_waste_facilities WHERE max_daily_capacity_ton <= 0 expected: 0 error_message: Unexpected rows detected in table suite: bronze diff --git a/dags/helpers/dbt_trino_operator.py b/dags/helpers/dbt_trino_operator.py index b96bd89..57765a9 100644 --- a/dags/helpers/dbt_trino_operator.py +++ b/dags/helpers/dbt_trino_operator.py @@ -73,10 +73,12 @@ class DbtTrinoIcebergOperator(BaseLakehouseOperator): return preferred candidates = [ + os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "dbt")), + "/opt/airflow/dag_bundles/merica-pipeline/tracking_repo/dbt", + "/opt/airflow/dag_bundles/merica-pipeline/dbt", "/opt/airflow/pipeline-bundles/dbt", "/opt/airflow/dag_bundles/dlh-pipeline-bundle/dbt", "/opt/airflow/dag_bundles/merica-pipeline-bundles/dbt", - "/opt/airflow/dag_bundles/merica-pipeline/dbt", os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "..", "..", "merica-pipeline", "dbt")), os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "..", "..", "merica-pipeline-bundles", "dbt")), "/opt/airflow/dbt/dlh_transformation", diff --git a/dags/helpers/sling_helper.py b/dags/helpers/sling_helper.py index fdd31c7..324d6d5 100644 --- a/dags/helpers/sling_helper.py +++ b/dags/helpers/sling_helper.py @@ -132,21 +132,30 @@ class SlingDBIngestionOperator(BaseOperator): def execute(self, context: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: cfg = self.bronze_config + source_cfg = cfg.get("source_config", {}) if "source_config" in cfg else cfg target_cfg = cfg.get("target", {}) bucket = target_cfg.get("bucket", "dlh-bronze") target_format = target_cfg.get("format", "parquet") - prefix = target_cfg.get("prefix", "db/permits") - table_name = cfg.get("table_name", "public.environmental_permits") - conn_id = cfg.get("connection_id", "conn_postgres_source_db") + + table_name = source_cfg.get("table_name") or source_cfg.get("table") or cfg.get("table_name") or "public.environmental_permits" + conn_id = source_cfg.get("conn_id") or source_cfg.get("connection_id") or cfg.get("connection_id", "conn_postgres_source_db") + clean_table = table_name.split(".")[-1] + + # Dynamic prefix resolution matching downstream DuckDB / Iceberg paths + if "ref_waste_facilities" in table_name or "ref_facilities" in table_name: + prefix = target_cfg.get("prefix") or "staging/ref_facilities" + elif "raw_waste_orders" in table_name or "waste_orders" in table_name: + prefix = target_cfg.get("prefix") or "staging/waste_orders" + else: + prefix = target_cfg.get("prefix") or f"staging/{clean_table}" exec_dt = datetime.utcnow() - clean_table = table_name.replace(".", "_") checkpoint_key = f"merica_checkpoint_{clean_table}" last_checkpoint = get_checkpoint_value(checkpoint_key, default="1970-01-01 00:00:00") self.log.info( - "Executing DB Ingestion on '%s' (mode: %s, last checkpoint: %s)", - table_name, self.ingestion_mode, last_checkpoint + "Executing DB Ingestion on '%s' (mode: %s, last checkpoint: %s, prefix: %s)", + table_name, self.ingestion_mode, last_checkpoint, prefix ) df = None @@ -178,10 +187,16 @@ class SlingDBIngestionOperator(BaseOperator): self.log.info("Running Native Database Extraction Engine.") db_url = resolve_db_connection_url(conn_id) try: - from sqlalchemy import create_engine + from sqlalchemy import create_engine, inspect engine = create_engine(db_url, connect_args={"connect_timeout": 5}) - if self.ingestion_mode == "incremental_watermark": + # Check whether watermark column exists in the table + insp = inspect(engine) + schema_name = table_name.split(".")[0] if "." in table_name else "public" + pure_tbl = table_name.split(".")[-1] + cols = [c["name"] for c in insp.get_columns(pure_tbl, schema=schema_name)] + + if self.ingestion_mode == "incremental_watermark" and self.watermark_column in cols: query = f"SELECT * FROM {table_name} WHERE {self.watermark_column} > '{last_checkpoint}' ORDER BY {self.watermark_column} ASC LIMIT {self.batch_size}" else: query = f"SELECT * FROM {table_name} LIMIT {self.batch_size}" @@ -217,6 +232,14 @@ class SlingDBIngestionOperator(BaseOperator): key=storage_key, target_format=target_format ) + # Also save flat key in prefix directory so glob patterns like s3://dlh-bronze/staging/ref_facilities/*.parquet match directly + flat_key = f"{prefix.strip('/')}/{filename}" + storage_manager.save_dataframe( + df=df, + bucket=bucket, + key=flat_key, + target_format=target_format + ) # 5. Sync to PostgreSQL Bronze CDC staging table for dbt-postgres medallion pipelines try: diff --git a/dbt/dbt_project.yml b/dbt/dbt_project.yml index 649c9cf..d8a8b3c 100644 --- a/dbt/dbt_project.yml +++ b/dbt/dbt_project.yml @@ -17,6 +17,9 @@ clean-targets: models: dlh_transformation: + bronze: + +materialized: view + +schema: bronze silver: +materialized: table +schema: silver diff --git a/dbt/models/bronze/bronze_ref_facilities.sql b/dbt/models/bronze/bronze_ref_facilities.sql new file mode 100644 index 0000000..1fa3b86 --- /dev/null +++ b/dbt/models/bronze/bronze_ref_facilities.sql @@ -0,0 +1,15 @@ +-- dbt Bronze Ref Facilities Model +{{ config(materialized='view') }} + +SELECT + facility_id, + facility_name, + operator_name, + license_number, + region_zone, + facility_type, + max_daily_capacity_ton, + facility_status, + compliance_rating, + created_at +FROM {{ source('public', 'ref_waste_facilities') }} diff --git a/dbt/models/bronze/bronze_waste_orders.sql b/dbt/models/bronze/bronze_waste_orders.sql new file mode 100644 index 0000000..ce9c79e --- /dev/null +++ b/dbt/models/bronze/bronze_waste_orders.sql @@ -0,0 +1,16 @@ +-- dbt Bronze Raw Waste Orders Model +{{ config(materialized='view') }} + +SELECT + order_id, + manifest_code, + facility_id, + generator_company, + waste_category, + volume_ton, + tariff_per_ton, + transport_fee, + order_status, + order_date, + created_at +FROM {{ source('public', 'raw_waste_orders') }} diff --git a/dbt/models/sources.yml b/dbt/models/sources.yml index 32985ef..4769568 100644 --- a/dbt/models/sources.yml +++ b/dbt/models/sources.yml @@ -17,3 +17,11 @@ sources: - name: environmental_permits_cdc - name: gws_waste_reports - name: spku_api_raw + + - name: public + schema: public + tables: + - name: ref_waste_facilities + description: "Master dataset fasilitas pengolahan limbah regional" + - name: raw_waste_orders + description: "Tabel transaksi manifest pengiriman dan penerimaan limbah"