From a5d766489adf73a6c8089730d42dfffd33ad2e69 Mon Sep 17 00:00:00 2001 From: syahril Date: Mon, 19 Oct 2026 01:53:47 +0700 Subject: [PATCH] feat(pipeline): add e2e orders master medallion flow and dbt models --- configs/e2e_orders_master_medallion_flow.yaml | 261 ++++++++++ .../gold/gold_regional_waste_metrics.sql | 16 + .../silver/silver_waste_manifest_enriched.sql | 27 + .../e2e_orders_master_medallion_flow.json | 481 ++++++++++++++++++ 4 files changed, 785 insertions(+) create mode 100644 configs/e2e_orders_master_medallion_flow.yaml create mode 100644 dbt/models/gold/gold_regional_waste_metrics.sql create mode 100644 dbt/models/silver/silver_waste_manifest_enriched.sql create mode 100644 metadata/e2e_orders_master_medallion_flow.json diff --git a/configs/e2e_orders_master_medallion_flow.yaml b/configs/e2e_orders_master_medallion_flow.yaml new file mode 100644 index 0000000..e6595ca --- /dev/null +++ b/configs/e2e_orders_master_medallion_flow.yaml @@ -0,0 +1,261 @@ +pipeline_id: e2e_orders_master_medallion_flow +description: Dynamic Data Lakehouse Pipeline via DLH Studio +schedule: 0 */1 * * * +mode: flexible +tasks: +- task_id: node-9826 + type: source_db + label: Extract Master Facilities + enabled: true + dependencies: [] + config: + conn: Postgres + conn_id: conn_postgres_source_db + table: public.ref_waste_facilities + table_name: public.ref_waste_facilities + target_type: minio_s3 +- task_id: node-0827 + type: duckdb_transform + label: DuckDB Master Facilities to Iceberg Bronze + enabled: true + dependencies: + - node-9826 + config: + layer: silver + source_parquet_path: s3://dlh-bronze/staging/ref_facilities/*.parquet + source_path: s3://dlh-bronze/staging/ref_facilities/*.parquet + target_table: bronze_ref_facilities + table_name: bronze_ref_facilities + primary_keys: facility_id + query: SELECT facility_id, facility_name, operator_name, license_number, region_zone, + facility_type, max_daily_capacity_ton, facility_status, compliance_rating, created_at + FROM read_parquet('s3://dlh-bronze/staging/ref_facilities/*.parquet') QUALIFY + ROW_NUMBER() OVER (PARTITION BY facility_id ORDER BY created_at DESC) = 1 + sql_query: SELECT facility_id, facility_name, operator_name, license_number, region_zone, + facility_type, max_daily_capacity_ton, facility_status, compliance_rating, created_at + FROM read_parquet('s3://dlh-bronze/staging/ref_facilities/*.parquet') QUALIFY + ROW_NUMBER() OVER (PARTITION BY facility_id ORDER BY created_at DESC) = 1 + transformation_sql: SELECT facility_id, facility_name, operator_name, license_number, + region_zone, facility_type, max_daily_capacity_ton, facility_status, compliance_rating, + created_at FROM read_parquet('s3://dlh-bronze/staging/ref_facilities/*.parquet') + QUALIFY ROW_NUMBER() OVER (PARTITION BY facility_id ORDER BY created_at DESC) + = 1 +- task_id: node-1812 + type: source_db + label: Extract Raw Waste Orders + enabled: true + dependencies: [] + config: + conn: Postgres + conn_id: conn_postgres_source_db + table: public.raw_waste_orders + table_name: public.raw_waste_orders + target_type: minio_s3 +- task_id: node-2810 + type: ge_quality + label: GE Orders Staging Quality Gate + enabled: true + dependencies: + - node-1812 + config: + ge_scope_level: all + ge_table_name: staging_waste_orders + test_table_row_count: true + test_table_schema_match: true + test_table_table_not_empty: true + table_min_rows: 1 + 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') + expected: 0 + error_message: Unexpected rows detected in table + suite: bronze + fail: Stop DAG +- task_id: node-3793 + type: duckdb_transform + label: DuckDB Orders to Iceberg Bronze + enabled: true + dependencies: + - node-2810 + config: + layer: silver + source_parquet_path: s3://dlh-bronze/staging/waste_orders/*.parquet + source_path: s3://dlh-bronze/staging/waste_orders/*.parquet + target_table: bronze_waste_orders + table_name: bronze_waste_orders + primary_keys: order_id + query: 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 read_parquet('s3://dlh-bronze/staging/waste_orders/*.parquet') QUALIFY + ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY created_at DESC) = 1 + sql_query: 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 read_parquet('s3://dlh-bronze/staging/waste_orders/*.parquet') QUALIFY + ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY created_at DESC) = 1 + transformation_sql: 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 read_parquet('s3://dlh-bronze/staging/waste_orders/*.parquet') + QUALIFY ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY created_at DESC) = + 1 +- task_id: node-4793 + type: ge_quality + label: GE Orders Bronze Iceberg Quality + enabled: true + dependencies: + - node-3793 + config: + ge_scope_level: all + ge_table_name: bronze_waste_orders + test_table_row_count: true + test_table_schema_match: true + test_table_table_not_empty: true + table_min_rows: 1 + 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') + expected: 0 + error_message: Unexpected rows detected in table + suite: bronze + fail: Stop DAG +- task_id: node-5809 + type: dbt_transform + label: dbt Silver Waste Manifest Enriched + enabled: true + dependencies: + - node-0827 + - node-4793 + config: + dbt_layer: silver + dbt_model: silver_waste_manifest_enriched + sel: silver_waste_manifest_enriched + tgt: postgres + trino_catalog: iceberg + sql_query_preview: "-- dbt Silver Cleansed Manifest Orders Enriched with Master\ + \ Facilities\nSELECT \n o.order_id,\n o.manifest_code,\n o.order_date,\n\ + \ o.generator_company,\n o.waste_category,\n o.volume_ton,\n o.tariff_per_ton,\n\ + \ o.transport_fee,\n ROUND((o.volume_ton * o.tariff_per_ton) + o.transport_fee,\ + \ 2) AS total_order_amount,\n o.order_status,\n f.facility_id,\n f.facility_name,\n\ + \ f.region_zone,\n f.facility_type,\n f.compliance_rating,\n f.facility_status,\n\ + \ CASE \n WHEN f.facility_status = 'ACTIVE' AND o.volume_ton <= f.max_daily_capacity_ton\ + \ THEN 'COMPLIANT'\n WHEN f.facility_status != 'ACTIVE' THEN 'UNLICENSED_FACILITY_WARNING'\n\ + \ ELSE 'CAPACITY_EXCEEDED_WARNING'\n END AS environmental_compliance_status\n\ + FROM {{ ref('bronze_waste_orders') }} o\nLEFT JOIN {{ ref('bronze_ref_facilities')\ + \ }} f\n ON o.facility_id = f.facility_id\nWHERE o.order_status != 'CANCELLED'" + query: "-- dbt Silver Cleansed Manifest Orders Enriched with Master Facilities\n\ + SELECT \n o.order_id,\n o.manifest_code,\n o.order_date,\n o.generator_company,\n\ + \ o.waste_category,\n o.volume_ton,\n o.tariff_per_ton,\n o.transport_fee,\n\ + \ ROUND((o.volume_ton * o.tariff_per_ton) + o.transport_fee, 2) AS total_order_amount,\n\ + \ o.order_status,\n f.facility_id,\n f.facility_name,\n f.region_zone,\n\ + \ f.facility_type,\n f.compliance_rating,\n f.facility_status,\n CASE \n\ + \ WHEN f.facility_status = 'ACTIVE' AND o.volume_ton <= f.max_daily_capacity_ton\ + \ THEN 'COMPLIANT'\n WHEN f.facility_status != 'ACTIVE' THEN 'UNLICENSED_FACILITY_WARNING'\n\ + \ ELSE 'CAPACITY_EXCEEDED_WARNING'\n END AS environmental_compliance_status\n\ + FROM {{ ref('bronze_waste_orders') }} o\nLEFT JOIN {{ ref('bronze_ref_facilities')\ + \ }} f\n ON o.facility_id = f.facility_id\nWHERE o.order_status != 'CANCELLED'" +- task_id: node-6793 + type: ge_quality + label: GE Silver Compliance Quality Gate + enabled: true + dependencies: + - node-5809 + config: + ge_scope_level: all + ge_table_name: silver_waste_manifest_enriched + test_table_row_count: true + test_table_schema_match: true + test_table_table_not_empty: true + table_min_rows: 1 + 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') + expected: 0 + error_message: Unexpected rows detected in table + suite: bronze + fail: Stop DAG +- task_id: node-7793 + type: dbt_transform + label: dbt Gold Regional Waste Revenue Mart + enabled: true + dependencies: + - node-6793 + config: + dbt_layer: gold + dbt_model: gold_regional_waste_metrics + sel: gold_regional_waste_metrics + tgt: postgres + trino_catalog: iceberg + sql_query_preview: "-- dbt Gold Regional Waste Revenue & Compliance Mart\nSELECT\ + \ \n region_zone,\n facility_type,\n waste_category,\n COUNT(order_id) AS\ + \ total_orders,\n COUNT(DISTINCT facility_id) AS active_facilities_count,\n\ + \ COUNT(DISTINCT generator_company) AS active_generators_count,\n ROUND(SUM(volume_ton),\ + \ 2) AS total_waste_volume_ton,\n ROUND(SUM(total_order_amount), 2) AS total_gross_revenue,\n\ + \ ROUND(AVG(total_order_amount / NULLIF(volume_ton, 0)), 2) AS avg_revenue_per_ton,\n\ + \ COUNT(CASE WHEN environmental_compliance_status = 'COMPLIANT' THEN 1 END)\ + \ AS compliant_orders_count,\n CURRENT_TIMESTAMP AS mart_refreshed_at\nFROM\ + \ {{ ref('silver_waste_manifest_enriched') }}\nGROUP BY region_zone, facility_type,\ + \ waste_category\nORDER BY total_waste_volume_ton DESC" + query: "-- dbt Gold Regional Waste Revenue & Compliance Mart\nSELECT \n region_zone,\n\ + \ facility_type,\n waste_category,\n COUNT(order_id) AS total_orders,\n \ + \ COUNT(DISTINCT facility_id) AS active_facilities_count,\n COUNT(DISTINCT\ + \ generator_company) AS active_generators_count,\n ROUND(SUM(volume_ton), 2)\ + \ AS total_waste_volume_ton,\n ROUND(SUM(total_order_amount), 2) AS total_gross_revenue,\n\ + \ ROUND(AVG(total_order_amount / NULLIF(volume_ton, 0)), 2) AS avg_revenue_per_ton,\n\ + \ COUNT(CASE WHEN environmental_compliance_status = 'COMPLIANT' THEN 1 END)\ + \ AS compliant_orders_count,\n CURRENT_TIMESTAMP AS mart_refreshed_at\nFROM\ + \ {{ ref('silver_waste_manifest_enriched') }}\nGROUP BY region_zone, facility_type,\ + \ waste_category\nORDER BY total_waste_volume_ton DESC" +- task_id: node-8871 + type: ge_quality + label: GE Gold Mart Integrity Gate + enabled: true + dependencies: + - node-7793 + config: + ge_scope_level: all + ge_table_name: gold_regional_waste_metrics + test_table_row_count: true + test_table_schema_match: true + test_table_table_not_empty: true + table_min_rows: 1 + 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') + expected: 0 + error_message: Unexpected rows detected in table + suite: bronze + fail: Stop DAG +- task_id: node-9855 + type: sitomat_callback + label: SITOMAT Governance Callback + enabled: true + dependencies: + - node-8871 + config: + layer: gold + artifact_ref: lakehouse.gold.gold_regional_waste_metrics + checksum: sha256-verified-waste-orders-pass + sitomat_base_url: http://localhost:3000 + openmetadata_fqn: dlh_airflow_service.pipeline.task + ge_status: PASSED + url: /sitomat/callback + stage: bronze diff --git a/dbt/models/gold/gold_regional_waste_metrics.sql b/dbt/models/gold/gold_regional_waste_metrics.sql new file mode 100644 index 0000000..226d139 --- /dev/null +++ b/dbt/models/gold/gold_regional_waste_metrics.sql @@ -0,0 +1,16 @@ +-- dbt Gold Regional Waste Revenue & Compliance Mart +SELECT + region_zone, + facility_type, + waste_category, + COUNT(order_id) AS total_orders, + COUNT(DISTINCT facility_id) AS active_facilities_count, + COUNT(DISTINCT generator_company) AS active_generators_count, + ROUND(SUM(volume_ton), 2) AS total_waste_volume_ton, + ROUND(SUM(total_order_amount), 2) AS total_gross_revenue, + ROUND(AVG(total_order_amount / NULLIF(volume_ton, 0)), 2) AS avg_revenue_per_ton, + COUNT(CASE WHEN environmental_compliance_status = 'COMPLIANT' THEN 1 END) AS compliant_orders_count, + CURRENT_TIMESTAMP AS mart_refreshed_at +FROM {{ ref('silver_waste_manifest_enriched') }} +GROUP BY region_zone, facility_type, waste_category +ORDER BY total_waste_volume_ton DESC \ No newline at end of file diff --git a/dbt/models/silver/silver_waste_manifest_enriched.sql b/dbt/models/silver/silver_waste_manifest_enriched.sql new file mode 100644 index 0000000..022785c --- /dev/null +++ b/dbt/models/silver/silver_waste_manifest_enriched.sql @@ -0,0 +1,27 @@ +-- dbt Silver Cleansed Manifest Orders Enriched with Master Facilities +SELECT + o.order_id, + o.manifest_code, + o.order_date, + o.generator_company, + o.waste_category, + o.volume_ton, + o.tariff_per_ton, + o.transport_fee, + ROUND((o.volume_ton * o.tariff_per_ton) + o.transport_fee, 2) AS total_order_amount, + o.order_status, + f.facility_id, + f.facility_name, + f.region_zone, + f.facility_type, + f.compliance_rating, + f.facility_status, + CASE + WHEN f.facility_status = 'ACTIVE' AND o.volume_ton <= f.max_daily_capacity_ton THEN 'COMPLIANT' + WHEN f.facility_status != 'ACTIVE' THEN 'UNLICENSED_FACILITY_WARNING' + ELSE 'CAPACITY_EXCEEDED_WARNING' + END AS environmental_compliance_status +FROM {{ ref('bronze_waste_orders') }} o +LEFT JOIN {{ ref('bronze_ref_facilities') }} f + ON o.facility_id = f.facility_id +WHERE o.order_status != 'CANCELLED' \ No newline at end of file diff --git a/metadata/e2e_orders_master_medallion_flow.json b/metadata/e2e_orders_master_medallion_flow.json new file mode 100644 index 0000000..75f06ad --- /dev/null +++ b/metadata/e2e_orders_master_medallion_flow.json @@ -0,0 +1,481 @@ +{ + "id": "e2e_orders_master_medallion_flow", + "description": "Dynamic Data Lakehouse Pipeline via DLH Studio", + "schedule": "0 */1 * * *", + "owner_id": "local-admin", + "owner_name": "Administrator Lokal (Break-Glass)", + "nodes": [ + { + "id": "node-9826", + "type": "customNode", + "position": { + "x": 80, + "y": 80 + }, + "data": { + "label": "Extract Master Facilities", + "type": "source_db", + "subtitle": "Postgres / MySQL / Oracle", + "isSource": true, + "configured": true, + "config": { + "conn": "Postgres", + "conn_id": "conn_postgres_source_db", + "table": "public.ref_waste_facilities", + "table_name": "public.ref_waste_facilities", + "target_type": "minio_s3" + } + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-0827", + "type": "customNode", + "position": { + "x": 270, + "y": 80 + }, + "data": { + "label": "DuckDB Master Facilities to Iceberg Bronze", + "type": "duckdb_transform", + "subtitle": "In-process SQL (<512MB)", + "configured": true, + "config": { + "layer": "silver", + "source_parquet_path": "s3://dlh-bronze/staging/ref_facilities/*.parquet", + "source_path": "s3://dlh-bronze/staging/ref_facilities/*.parquet", + "target_table": "bronze_ref_facilities", + "table_name": "bronze_ref_facilities", + "primary_keys": "facility_id", + "query": "SELECT facility_id, facility_name, operator_name, license_number, region_zone, facility_type, max_daily_capacity_ton, facility_status, compliance_rating, created_at FROM read_parquet('s3://dlh-bronze/staging/ref_facilities/*.parquet') QUALIFY ROW_NUMBER() OVER (PARTITION BY facility_id ORDER BY created_at DESC) = 1", + "sql_query": "SELECT facility_id, facility_name, operator_name, license_number, region_zone, facility_type, max_daily_capacity_ton, facility_status, compliance_rating, created_at FROM read_parquet('s3://dlh-bronze/staging/ref_facilities/*.parquet') QUALIFY ROW_NUMBER() OVER (PARTITION BY facility_id ORDER BY created_at DESC) = 1", + "transformation_sql": "SELECT facility_id, facility_name, operator_name, license_number, region_zone, facility_type, max_daily_capacity_ton, facility_status, compliance_rating, created_at FROM read_parquet('s3://dlh-bronze/staging/ref_facilities/*.parquet') QUALIFY ROW_NUMBER() OVER (PARTITION BY facility_id ORDER BY created_at DESC) = 1" + } + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-1812", + "type": "customNode", + "position": { + "x": 460, + "y": 80 + }, + "data": { + "label": "Extract Raw Waste Orders", + "type": "source_db", + "subtitle": "Postgres / MySQL / Oracle", + "isSource": true, + "configured": true, + "config": { + "conn": "Postgres", + "conn_id": "conn_postgres_source_db", + "table": "public.raw_waste_orders", + "table_name": "public.raw_waste_orders", + "target_type": "minio_s3" + } + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-2810", + "type": "customNode", + "position": { + "x": 650, + "y": 80 + }, + "data": { + "label": "GE Orders Staging Quality Gate", + "type": "ge_quality", + "subtitle": "Great Expectations", + "configured": true, + "config": { + "ge_scope_level": "all", + "ge_table_name": "staging_waste_orders", + "test_table_row_count": true, + "test_table_schema_match": true, + "test_table_table_not_empty": true, + "table_min_rows": 1, + "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')", + "expected": 0, + "error_message": "Unexpected rows detected in table", + "suite": "bronze", + "fail": "Stop DAG" + } + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-3793", + "type": "customNode", + "position": { + "x": 840, + "y": 80 + }, + "data": { + "label": "DuckDB Orders to Iceberg Bronze", + "type": "duckdb_transform", + "subtitle": "In-process SQL (<512MB)", + "configured": true, + "config": { + "layer": "silver", + "source_parquet_path": "s3://dlh-bronze/staging/waste_orders/*.parquet", + "source_path": "s3://dlh-bronze/staging/waste_orders/*.parquet", + "target_table": "bronze_waste_orders", + "table_name": "bronze_waste_orders", + "primary_keys": "order_id", + "query": "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 read_parquet('s3://dlh-bronze/staging/waste_orders/*.parquet') QUALIFY ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY created_at DESC) = 1", + "sql_query": "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 read_parquet('s3://dlh-bronze/staging/waste_orders/*.parquet') QUALIFY ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY created_at DESC) = 1", + "transformation_sql": "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 read_parquet('s3://dlh-bronze/staging/waste_orders/*.parquet') QUALIFY ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY created_at DESC) = 1" + } + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-4793", + "type": "customNode", + "position": { + "x": 1030, + "y": 80 + }, + "data": { + "label": "GE Orders Bronze Iceberg Quality", + "type": "ge_quality", + "subtitle": "Great Expectations", + "configured": true, + "config": { + "ge_scope_level": "all", + "ge_table_name": "bronze_waste_orders", + "test_table_row_count": true, + "test_table_schema_match": true, + "test_table_table_not_empty": true, + "table_min_rows": 1, + "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')", + "expected": 0, + "error_message": "Unexpected rows detected in table", + "suite": "bronze", + "fail": "Stop DAG" + } + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-5809", + "type": "customNode", + "position": { + "x": 1220, + "y": 80 + }, + "data": { + "label": "dbt Silver Waste Manifest Enriched", + "type": "dbt_transform", + "subtitle": "dbt-postgres", + "configured": true, + "config": { + "dbt_layer": "silver", + "dbt_model": "silver_waste_manifest_enriched", + "sel": "silver_waste_manifest_enriched", + "tgt": "postgres", + "trino_catalog": "iceberg", + "sql_query_preview": "-- dbt Silver Cleansed Manifest Orders Enriched with Master Facilities\nSELECT \n o.order_id,\n o.manifest_code,\n o.order_date,\n o.generator_company,\n o.waste_category,\n o.volume_ton,\n o.tariff_per_ton,\n o.transport_fee,\n ROUND((o.volume_ton * o.tariff_per_ton) + o.transport_fee, 2) AS total_order_amount,\n o.order_status,\n f.facility_id,\n f.facility_name,\n f.region_zone,\n f.facility_type,\n f.compliance_rating,\n f.facility_status,\n CASE \n WHEN f.facility_status = 'ACTIVE' AND o.volume_ton <= f.max_daily_capacity_ton THEN 'COMPLIANT'\n WHEN f.facility_status != 'ACTIVE' THEN 'UNLICENSED_FACILITY_WARNING'\n ELSE 'CAPACITY_EXCEEDED_WARNING'\n END AS environmental_compliance_status\nFROM {{ ref('bronze_waste_orders') }} o\nLEFT JOIN {{ ref('bronze_ref_facilities') }} f\n ON o.facility_id = f.facility_id\nWHERE o.order_status != 'CANCELLED'", + "query": "-- dbt Silver Cleansed Manifest Orders Enriched with Master Facilities\nSELECT \n o.order_id,\n o.manifest_code,\n o.order_date,\n o.generator_company,\n o.waste_category,\n o.volume_ton,\n o.tariff_per_ton,\n o.transport_fee,\n ROUND((o.volume_ton * o.tariff_per_ton) + o.transport_fee, 2) AS total_order_amount,\n o.order_status,\n f.facility_id,\n f.facility_name,\n f.region_zone,\n f.facility_type,\n f.compliance_rating,\n f.facility_status,\n CASE \n WHEN f.facility_status = 'ACTIVE' AND o.volume_ton <= f.max_daily_capacity_ton THEN 'COMPLIANT'\n WHEN f.facility_status != 'ACTIVE' THEN 'UNLICENSED_FACILITY_WARNING'\n ELSE 'CAPACITY_EXCEEDED_WARNING'\n END AS environmental_compliance_status\nFROM {{ ref('bronze_waste_orders') }} o\nLEFT JOIN {{ ref('bronze_ref_facilities') }} f\n ON o.facility_id = f.facility_id\nWHERE o.order_status != 'CANCELLED'" + } + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-6793", + "type": "customNode", + "position": { + "x": 1410, + "y": 80 + }, + "data": { + "label": "GE Silver Compliance Quality Gate", + "type": "ge_quality", + "subtitle": "Great Expectations", + "configured": true, + "config": { + "ge_scope_level": "all", + "ge_table_name": "silver_waste_manifest_enriched", + "test_table_row_count": true, + "test_table_schema_match": true, + "test_table_table_not_empty": true, + "table_min_rows": 1, + "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')", + "expected": 0, + "error_message": "Unexpected rows detected in table", + "suite": "bronze", + "fail": "Stop DAG" + } + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-7793", + "type": "customNode", + "position": { + "x": 1600, + "y": 80 + }, + "data": { + "label": "dbt Gold Regional Waste Revenue Mart", + "type": "dbt_transform", + "subtitle": "dbt-postgres", + "configured": true, + "config": { + "dbt_layer": "gold", + "dbt_model": "gold_regional_waste_metrics", + "sel": "gold_regional_waste_metrics", + "tgt": "postgres", + "trino_catalog": "iceberg", + "sql_query_preview": "-- dbt Gold Regional Waste Revenue & Compliance Mart\nSELECT \n region_zone,\n facility_type,\n waste_category,\n COUNT(order_id) AS total_orders,\n COUNT(DISTINCT facility_id) AS active_facilities_count,\n COUNT(DISTINCT generator_company) AS active_generators_count,\n ROUND(SUM(volume_ton), 2) AS total_waste_volume_ton,\n ROUND(SUM(total_order_amount), 2) AS total_gross_revenue,\n ROUND(AVG(total_order_amount / NULLIF(volume_ton, 0)), 2) AS avg_revenue_per_ton,\n COUNT(CASE WHEN environmental_compliance_status = 'COMPLIANT' THEN 1 END) AS compliant_orders_count,\n CURRENT_TIMESTAMP AS mart_refreshed_at\nFROM {{ ref('silver_waste_manifest_enriched') }}\nGROUP BY region_zone, facility_type, waste_category\nORDER BY total_waste_volume_ton DESC", + "query": "-- dbt Gold Regional Waste Revenue & Compliance Mart\nSELECT \n region_zone,\n facility_type,\n waste_category,\n COUNT(order_id) AS total_orders,\n COUNT(DISTINCT facility_id) AS active_facilities_count,\n COUNT(DISTINCT generator_company) AS active_generators_count,\n ROUND(SUM(volume_ton), 2) AS total_waste_volume_ton,\n ROUND(SUM(total_order_amount), 2) AS total_gross_revenue,\n ROUND(AVG(total_order_amount / NULLIF(volume_ton, 0)), 2) AS avg_revenue_per_ton,\n COUNT(CASE WHEN environmental_compliance_status = 'COMPLIANT' THEN 1 END) AS compliant_orders_count,\n CURRENT_TIMESTAMP AS mart_refreshed_at\nFROM {{ ref('silver_waste_manifest_enriched') }}\nGROUP BY region_zone, facility_type, waste_category\nORDER BY total_waste_volume_ton DESC" + } + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-8871", + "type": "customNode", + "position": { + "x": 1790, + "y": 80 + }, + "data": { + "label": "GE Gold Mart Integrity Gate", + "type": "ge_quality", + "subtitle": "Great Expectations", + "configured": true, + "config": { + "ge_scope_level": "all", + "ge_table_name": "gold_regional_waste_metrics", + "test_table_row_count": true, + "test_table_schema_match": true, + "test_table_table_not_empty": true, + "table_min_rows": 1, + "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')", + "expected": 0, + "error_message": "Unexpected rows detected in table", + "suite": "bronze", + "fail": "Stop DAG" + } + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-9855", + "type": "customNode", + "position": { + "x": 1980, + "y": 80 + }, + "data": { + "label": "SITOMAT Governance Callback", + "type": "sitomat_callback", + "subtitle": "Status callback", + "configured": true, + "config": { + "layer": "gold", + "artifact_ref": "lakehouse.gold.gold_regional_waste_metrics", + "checksum": "sha256-verified-waste-orders-pass", + "sitomat_base_url": "http://localhost:3000", + "openmetadata_fqn": "dlh_airflow_service.pipeline.task", + "ge_status": "PASSED", + "url": "/sitomat/callback", + "stage": "bronze" + } + }, + "width": 96, + "height": 96, + "selected": true, + "dragging": false + } + ], + "edges": [ + { + "source": "node-9826", + "sourceHandle": null, + "target": "node-0827", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-9826-node-0827" + }, + { + "source": "node-0827", + "sourceHandle": null, + "target": "node-5809", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-0827-node-5809" + }, + { + "source": "node-1812", + "sourceHandle": null, + "target": "node-2810", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-1812-node-2810" + }, + { + "source": "node-2810", + "sourceHandle": null, + "target": "node-3793", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-2810-node-3793" + }, + { + "source": "node-3793", + "sourceHandle": null, + "target": "node-4793", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-3793-node-4793" + }, + { + "source": "node-4793", + "sourceHandle": null, + "target": "node-5809", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-4793-node-5809" + }, + { + "source": "node-5809", + "sourceHandle": null, + "target": "node-6793", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-5809-node-6793" + }, + { + "source": "node-6793", + "sourceHandle": null, + "target": "node-7793", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-6793-node-7793" + }, + { + "source": "node-7793", + "sourceHandle": null, + "target": "node-8871", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-7793-node-8871" + }, + { + "source": "node-8871", + "sourceHandle": null, + "target": "node-9855", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-8871-node-9855" + } + ], + "notes": [], + "is_draft": false, + "status": "deployed", + "updated_at": "2026-10-18T18:28:07.599927Z" +} \ No newline at end of file