From b37ffc52ae0af7da035c97b21c0f0582e7dd9d0b Mon Sep 17 00:00:00 2001 From: syahril Date: Mon, 19 Oct 2026 03:50:17 +0700 Subject: [PATCH] feat(pipeline): add e2e cems industrial emissions medallion pipeline and dbt models --- .../e2e_cems_industrial_emissions_flow.yaml | 279 ++++++++++ ...gold_industry_emission_compliance_mart.sql | 21 + .../silver/silver_cems_emission_enriched.sql | 31 ++ dbt/models/sources.yml | 54 +- .../e2e_cems_industrial_emissions_flow.json | 481 ++++++++++++++++++ metadata/rbac_mappings.json | 2 +- 6 files changed, 842 insertions(+), 26 deletions(-) create mode 100644 configs/e2e_cems_industrial_emissions_flow.yaml create mode 100644 dbt/models/gold/gold_industry_emission_compliance_mart.sql create mode 100644 dbt/models/silver/silver_cems_emission_enriched.sql create mode 100644 metadata/e2e_cems_industrial_emissions_flow.json diff --git a/configs/e2e_cems_industrial_emissions_flow.yaml b/configs/e2e_cems_industrial_emissions_flow.yaml new file mode 100644 index 0000000..3ddae55 --- /dev/null +++ b/configs/e2e_cems_industrial_emissions_flow.yaml @@ -0,0 +1,279 @@ +pipeline_id: e2e_cems_industrial_emissions_flow +description: Dynamic Data Lakehouse Pipeline via DLH Studio +schedule: 0 */1 * * * +mode: flexible +governance: + domain: Lingkungan Hidup & Kehutanan + owner: de_team_dlh + tier: Tier.Tier1 + certification: Certification.Gold +tasks: +- task_id: node-1975 + type: source_db + label: Extract Master Emission Sources + enabled: true + dependencies: [] + config: + conn: Postgres + conn_id: conn_postgres_source_db + table: public.ref_emission_sources + table_name: public.ref_emission_sources + target_type: minio_s3 +- task_id: node-3042 + type: duckdb_transform + label: DuckDB Master to Bronze Iceberg + enabled: true + dependencies: + - node-1975 + config: + layer: silver + source_parquet_path: s3://dlh-bronze/staging/ref_emission_sources/*.parquet + source_path: s3://dlh-bronze/staging/ref_emission_sources/*.parquet + target_table: bronze_ref_emission_sources + table_name: bronze_ref_emission_sources + primary_keys: source_id + query: SELECT source_id, facility_name, industry_sector, chimney_code, chimney_height_m, + max_so2_limit_mg_nm3, max_nox_limit_mg_nm3, max_pm_limit_mg_nm3, region_zone, + license_status, created_at FROM read_parquet('s3://dlh-bronze/staging/ref_emission_sources/*.parquet') + QUALIFY ROW_NUMBER() OVER (PARTITION BY source_id ORDER BY created_at DESC) + = 1 + sql_query: SELECT source_id, facility_name, industry_sector, chimney_code, chimney_height_m, + max_so2_limit_mg_nm3, max_nox_limit_mg_nm3, max_pm_limit_mg_nm3, region_zone, + license_status, created_at FROM read_parquet('s3://dlh-bronze/staging/ref_emission_sources/*.parquet') + QUALIFY ROW_NUMBER() OVER (PARTITION BY source_id ORDER BY created_at DESC) + = 1 + transformation_sql: SELECT source_id, facility_name, industry_sector, chimney_code, + chimney_height_m, max_so2_limit_mg_nm3, max_nox_limit_mg_nm3, max_pm_limit_mg_nm3, + region_zone, license_status, created_at FROM read_parquet('s3://dlh-bronze/staging/ref_emission_sources/*.parquet') + QUALIFY ROW_NUMBER() OVER (PARTITION BY source_id ORDER BY created_at DESC) + = 1 +- task_id: node-4059 + type: source_db + label: Extract Raw CEMS Telemetry + enabled: true + dependencies: [] + config: + conn: Postgres + conn_id: conn_postgres_source_db + table: public.raw_cems_telemetry_logs + table_name: public.raw_cems_telemetry_logs + target_type: minio_s3 +- task_id: node-5075 + type: ge_quality + label: GE Telemetry Staging Check + enabled: true + dependencies: + - node-4059 + config: + ge_scope_level: all + ge_table_name: staging_cems_telemetry + 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-6110 + type: duckdb_transform + label: DuckDB Telemetry to Bronze Iceberg + enabled: true + dependencies: + - node-5075 + config: + layer: silver + source_parquet_path: s3://dlh-bronze/staging/raw_cems_telemetry/*.parquet + source_path: s3://dlh-bronze/staging/raw_cems_telemetry/*.parquet + target_table: bronze_cems_telemetry_logs + table_name: bronze_cems_telemetry_logs + primary_keys: log_id + query: SELECT log_id, source_id, telemetry_timestamp, so2_mg_nm3, nox_mg_nm3, + pm_dust_mg_nm3, co_mg_nm3, flue_gas_flow_m3_s, flue_temp_celsius, sensor_status, + created_at FROM read_parquet('s3://dlh-bronze/staging/raw_cems_telemetry/*.parquet') + QUALIFY ROW_NUMBER() OVER (PARTITION BY log_id ORDER BY created_at DESC) = 1 + sql_query: SELECT log_id, source_id, telemetry_timestamp, so2_mg_nm3, nox_mg_nm3, + pm_dust_mg_nm3, co_mg_nm3, flue_gas_flow_m3_s, flue_temp_celsius, sensor_status, + created_at FROM read_parquet('s3://dlh-bronze/staging/raw_cems_telemetry/*.parquet') + QUALIFY ROW_NUMBER() OVER (PARTITION BY log_id ORDER BY created_at DESC) = 1 + transformation_sql: SELECT log_id, source_id, telemetry_timestamp, so2_mg_nm3, + nox_mg_nm3, pm_dust_mg_nm3, co_mg_nm3, flue_gas_flow_m3_s, flue_temp_celsius, + sensor_status, created_at FROM read_parquet('s3://dlh-bronze/staging/raw_cems_telemetry/*.parquet') + QUALIFY ROW_NUMBER() OVER (PARTITION BY log_id ORDER BY created_at DESC) = 1 +- task_id: node-7127 + type: ge_quality + label: GE Telemetry Bronze Quality + enabled: true + dependencies: + - node-6110 + config: + ge_scope_level: all + ge_table_name: bronze_cems_telemetry_logs + 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-8143 + type: dbt_transform + label: dbt Silver CEMS Enriched + enabled: true + dependencies: + - node-3042 + - node-7127 + config: + dbt_layer: silver + dbt_model: silver_cems_emission_enriched + sel: silver_cems_emission_enriched + tgt: postgres + trino_catalog: iceberg + sql_query_preview: "-- dbt Silver Cleansed CEMS Telemetry Enriched with Master\ + \ Sources\nSELECT \n t.log_id,\n t.telemetry_timestamp,\n t.source_id,\n\ + \ s.facility_name,\n s.industry_sector,\n s.chimney_code,\n s.region_zone,\n\ + \ t.so2_mg_nm3,\n s.max_so2_limit_mg_nm3,\n t.nox_mg_nm3,\n s.max_nox_limit_mg_nm3,\n\ + \ t.pm_dust_mg_nm3,\n s.max_pm_limit_mg_nm3,\n t.co_mg_nm3,\n t.flue_gas_flow_m3_s,\n\ + \ t.flue_temp_celsius,\n ROUND((t.so2_mg_nm3 * t.flue_gas_flow_m3_s * 3600.0)\ + \ / 1000000.0, 3) AS so2_mass_rate_kg_h,\n ROUND((t.nox_mg_nm3 * t.flue_gas_flow_m3_s\ + \ * 3600.0) / 1000000.0, 3) AS nox_mass_rate_kg_h,\n ROUND((t.pm_dust_mg_nm3\ + \ * t.flue_gas_flow_m3_s * 3600.0) / 1000000.0, 3) AS pm_mass_rate_kg_h,\n \ + \ CASE \n WHEN t.so2_mg_nm3 > s.max_so2_limit_mg_nm3 OR t.nox_mg_nm3 > s.max_nox_limit_mg_nm3\ + \ OR t.pm_dust_mg_nm3 > s.max_pm_limit_mg_nm3 \n THEN 'EXCEEDING_LIMIT'\n\ + \ ELSE 'COMPLIANT'\n END AS compliance_status,\n t.sensor_status\nFROM\ + \ {{ ref('bronze_cems_telemetry_logs') }} t\nJOIN {{ ref('bronze_ref_emission_sources')\ + \ }} s\n ON t.source_id = s.source_id\nWHERE t.sensor_status = 'VALID'" + query: "-- dbt Silver Cleansed CEMS Telemetry Enriched with Master Sources\nSELECT\ + \ \n t.log_id,\n t.telemetry_timestamp,\n t.source_id,\n s.facility_name,\n\ + \ s.industry_sector,\n s.chimney_code,\n s.region_zone,\n t.so2_mg_nm3,\n\ + \ s.max_so2_limit_mg_nm3,\n t.nox_mg_nm3,\n s.max_nox_limit_mg_nm3,\n t.pm_dust_mg_nm3,\n\ + \ s.max_pm_limit_mg_nm3,\n t.co_mg_nm3,\n t.flue_gas_flow_m3_s,\n t.flue_temp_celsius,\n\ + \ ROUND((t.so2_mg_nm3 * t.flue_gas_flow_m3_s * 3600.0) / 1000000.0, 3) AS so2_mass_rate_kg_h,\n\ + \ ROUND((t.nox_mg_nm3 * t.flue_gas_flow_m3_s * 3600.0) / 1000000.0, 3) AS nox_mass_rate_kg_h,\n\ + \ ROUND((t.pm_dust_mg_nm3 * t.flue_gas_flow_m3_s * 3600.0) / 1000000.0, 3)\ + \ AS pm_mass_rate_kg_h,\n CASE \n WHEN t.so2_mg_nm3 > s.max_so2_limit_mg_nm3\ + \ OR t.nox_mg_nm3 > s.max_nox_limit_mg_nm3 OR t.pm_dust_mg_nm3 > s.max_pm_limit_mg_nm3\ + \ \n THEN 'EXCEEDING_LIMIT'\n ELSE 'COMPLIANT'\n END AS compliance_status,\n\ + \ t.sensor_status\nFROM {{ ref('bronze_cems_telemetry_logs') }} t\nJOIN {{\ + \ ref('bronze_ref_emission_sources') }} s\n ON t.source_id = s.source_id\n\ + WHERE t.sensor_status = 'VALID'" +- task_id: node-9175 + type: ge_quality + label: GE Silver Compliance Quality Gate + enabled: true + dependencies: + - node-8143 + config: + ge_scope_level: all + ge_table_name: silver_cems_emission_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-0193 + type: dbt_transform + label: dbt Gold Industrial Compliance Mart + enabled: true + dependencies: + - node-9175 + config: + dbt_layer: gold + dbt_model: gold_industry_emission_compliance_mart + sel: gold_industry_emission_compliance_mart + tgt: postgres + trino_catalog: iceberg + sql_query_preview: "-- dbt Gold Industrial Emission Compliance & Load Mart\nSELECT\ + \ \n region_zone,\n industry_sector,\n COUNT(log_id) AS total_telemetry_events,\n\ + \ COUNT(DISTINCT source_id) AS active_chimneys_count,\n ROUND(AVG(so2_mg_nm3),\ + \ 2) AS avg_so2_concentration,\n ROUND(AVG(nox_mg_nm3), 2) AS avg_nox_concentration,\n\ + \ ROUND(AVG(pm_dust_mg_nm3), 2) AS avg_pm_concentration,\n ROUND(SUM(so2_mass_rate_kg_h)\ + \ / 1000.0, 3) AS total_so2_emission_ton,\n ROUND(SUM(nox_mass_rate_kg_h) /\ + \ 1000.0, 3) AS total_nox_emission_ton,\n ROUND(SUM(pm_mass_rate_kg_h) / 1000.0,\ + \ 3) AS total_pm_emission_ton,\n COUNT(CASE WHEN compliance_status = 'EXCEEDING_LIMIT'\ + \ THEN 1 END) AS total_violations_count,\n ROUND(\n (COUNT(CASE WHEN compliance_status\ + \ = 'COMPLIANT' THEN 1 END)::NUMERIC / NULLIF(COUNT(log_id), 0)) * 100.0,\n\ + \ 2\n ) AS compliance_rate_percent,\n CURRENT_TIMESTAMP AS mart_refreshed_at\n\ + FROM {{ ref('silver_cems_emission_enriched') }}\nGROUP BY region_zone, industry_sector\n\ + ORDER BY total_violations_count DESC, total_so2_emission_ton DESC" + query: "-- dbt Gold Industrial Emission Compliance & Load Mart\nSELECT \n region_zone,\n\ + \ industry_sector,\n COUNT(log_id) AS total_telemetry_events,\n COUNT(DISTINCT\ + \ source_id) AS active_chimneys_count,\n ROUND(AVG(so2_mg_nm3), 2) AS avg_so2_concentration,\n\ + \ ROUND(AVG(nox_mg_nm3), 2) AS avg_nox_concentration,\n ROUND(AVG(pm_dust_mg_nm3),\ + \ 2) AS avg_pm_concentration,\n ROUND(SUM(so2_mass_rate_kg_h) / 1000.0, 3)\ + \ AS total_so2_emission_ton,\n ROUND(SUM(nox_mass_rate_kg_h) / 1000.0, 3) AS\ + \ total_nox_emission_ton,\n ROUND(SUM(pm_mass_rate_kg_h) / 1000.0, 3) AS total_pm_emission_ton,\n\ + \ COUNT(CASE WHEN compliance_status = 'EXCEEDING_LIMIT' THEN 1 END) AS total_violations_count,\n\ + \ ROUND(\n (COUNT(CASE WHEN compliance_status = 'COMPLIANT' THEN 1 END)::NUMERIC\ + \ / NULLIF(COUNT(log_id), 0)) * 100.0,\n 2\n ) AS compliance_rate_percent,\n\ + \ CURRENT_TIMESTAMP AS mart_refreshed_at\nFROM {{ ref('silver_cems_emission_enriched')\ + \ }}\nGROUP BY region_zone, industry_sector\nORDER BY total_violations_count\ + \ DESC, total_so2_emission_ton DESC" +- task_id: node-1241 + type: ge_quality + label: GE Gold Compliance Mart Gate + enabled: true + dependencies: + - node-0193 + config: + ge_scope_level: all + ge_table_name: gold_industry_emission_compliance_mart + 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-2409 + type: sitomat_callback + label: SITOMAT Governance Callback + enabled: true + dependencies: + - node-1241 + config: + layer: gold + artifact_ref: lakehouse.gold.gold_industry_emission_compliance_mart + checksum: sha256-verified-cems-emissions-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_industry_emission_compliance_mart.sql b/dbt/models/gold/gold_industry_emission_compliance_mart.sql new file mode 100644 index 0000000..14334aa --- /dev/null +++ b/dbt/models/gold/gold_industry_emission_compliance_mart.sql @@ -0,0 +1,21 @@ +-- dbt Gold Industrial Emission Compliance & Load Mart +SELECT + region_zone, + industry_sector, + COUNT(log_id) AS total_telemetry_events, + COUNT(DISTINCT source_id) AS active_chimneys_count, + ROUND(AVG(so2_mg_nm3), 2) AS avg_so2_concentration, + ROUND(AVG(nox_mg_nm3), 2) AS avg_nox_concentration, + ROUND(AVG(pm_dust_mg_nm3), 2) AS avg_pm_concentration, + ROUND(SUM(so2_mass_rate_kg_h) / 1000.0, 3) AS total_so2_emission_ton, + ROUND(SUM(nox_mass_rate_kg_h) / 1000.0, 3) AS total_nox_emission_ton, + ROUND(SUM(pm_mass_rate_kg_h) / 1000.0, 3) AS total_pm_emission_ton, + COUNT(CASE WHEN compliance_status = 'EXCEEDING_LIMIT' THEN 1 END) AS total_violations_count, + ROUND( + (COUNT(CASE WHEN compliance_status = 'COMPLIANT' THEN 1 END)::NUMERIC / NULLIF(COUNT(log_id), 0)) * 100.0, + 2 + ) AS compliance_rate_percent, + CURRENT_TIMESTAMP AS mart_refreshed_at +FROM {{ ref('silver_cems_emission_enriched') }} +GROUP BY region_zone, industry_sector +ORDER BY total_violations_count DESC, total_so2_emission_ton DESC \ No newline at end of file diff --git a/dbt/models/silver/silver_cems_emission_enriched.sql b/dbt/models/silver/silver_cems_emission_enriched.sql new file mode 100644 index 0000000..793f874 --- /dev/null +++ b/dbt/models/silver/silver_cems_emission_enriched.sql @@ -0,0 +1,31 @@ +-- dbt Silver Cleansed CEMS Telemetry Enriched with Master Sources +SELECT + t.log_id, + t.telemetry_timestamp, + t.source_id, + s.facility_name, + s.industry_sector, + s.chimney_code, + s.region_zone, + t.so2_mg_nm3, + s.max_so2_limit_mg_nm3, + t.nox_mg_nm3, + s.max_nox_limit_mg_nm3, + t.pm_dust_mg_nm3, + s.max_pm_limit_mg_nm3, + t.co_mg_nm3, + t.flue_gas_flow_m3_s, + t.flue_temp_celsius, + ROUND((t.so2_mg_nm3 * t.flue_gas_flow_m3_s * 3600.0) / 1000000.0, 3) AS so2_mass_rate_kg_h, + ROUND((t.nox_mg_nm3 * t.flue_gas_flow_m3_s * 3600.0) / 1000000.0, 3) AS nox_mass_rate_kg_h, + ROUND((t.pm_dust_mg_nm3 * t.flue_gas_flow_m3_s * 3600.0) / 1000000.0, 3) AS pm_mass_rate_kg_h, + CASE + WHEN t.so2_mg_nm3 > s.max_so2_limit_mg_nm3 OR t.nox_mg_nm3 > s.max_nox_limit_mg_nm3 OR t.pm_dust_mg_nm3 > s.max_pm_limit_mg_nm3 + THEN 'EXCEEDING_LIMIT' + ELSE 'COMPLIANT' + END AS compliance_status, + t.sensor_status +FROM {{ ref('bronze_cems_telemetry_logs') }} t +JOIN {{ ref('bronze_ref_emission_sources') }} s + ON t.source_id = s.source_id +WHERE t.sensor_status = 'VALID' \ No newline at end of file diff --git a/dbt/models/sources.yml b/dbt/models/sources.yml index 4769568..baeeb43 100644 --- a/dbt/models/sources.yml +++ b/dbt/models/sources.yml @@ -1,27 +1,31 @@ version: 2 - sources: - - name: bronze_lakehouse - schema: bronze_lakehouse - tables: - - name: environmental_permits_cdc - description: "Tabel landing Bronze staging untuk Change Data Capture (CDC) Perizinan Lingkungan" - - name: gws_waste_reports - description: "Tabel landing Bronze staging laporan penanganan sampah Google Workspace" - - name: spku_api_raw - description: "Tabel landing Bronze staging raw JSON payload sensor kualitas udara SPKU" - - - name: bronze - schema: bronze_lakehouse - tables: - - 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" +- name: bronze_lakehouse + schema: bronze_lakehouse + tables: + - name: environmental_permits_cdc + description: Tabel landing Bronze staging untuk Change Data Capture (CDC) Perizinan + Lingkungan + - name: gws_waste_reports + description: Tabel landing Bronze staging laporan penanganan sampah Google Workspace + - name: spku_api_raw + description: Tabel landing Bronze staging raw JSON payload sensor kualitas udara + SPKU +- name: bronze + schema: bronze_lakehouse + tables: + - 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 + - name: ref_emission_sources + description: Auto-registered source table 'ref_emission_sources' from canvas pipeline + - name: raw_cems_telemetry_logs + description: Auto-registered source table 'raw_cems_telemetry_logs' from canvas + pipeline diff --git a/metadata/e2e_cems_industrial_emissions_flow.json b/metadata/e2e_cems_industrial_emissions_flow.json new file mode 100644 index 0000000..001d87e --- /dev/null +++ b/metadata/e2e_cems_industrial_emissions_flow.json @@ -0,0 +1,481 @@ +{ + "id": "e2e_cems_industrial_emissions_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-1975", + "type": "customNode", + "position": { + "x": 80, + "y": 80 + }, + "data": { + "label": "Extract Master Emission Sources", + "type": "source_db", + "subtitle": "Postgres / MySQL / Oracle", + "isSource": true, + "configured": true, + "config": { + "conn": "Postgres", + "conn_id": "conn_postgres_source_db", + "table": "public.ref_emission_sources", + "table_name": "public.ref_emission_sources", + "target_type": "minio_s3" + } + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-3042", + "type": "customNode", + "position": { + "x": 270, + "y": 80 + }, + "data": { + "label": "DuckDB Master to Bronze Iceberg", + "type": "duckdb_transform", + "subtitle": "In-process SQL (<512MB)", + "configured": true, + "config": { + "layer": "silver", + "source_parquet_path": "s3://dlh-bronze/staging/ref_emission_sources/*.parquet", + "source_path": "s3://dlh-bronze/staging/ref_emission_sources/*.parquet", + "target_table": "bronze_ref_emission_sources", + "table_name": "bronze_ref_emission_sources", + "primary_keys": "source_id", + "query": "SELECT source_id, facility_name, industry_sector, chimney_code, chimney_height_m, max_so2_limit_mg_nm3, max_nox_limit_mg_nm3, max_pm_limit_mg_nm3, region_zone, license_status, created_at FROM read_parquet('s3://dlh-bronze/staging/ref_emission_sources/*.parquet') QUALIFY ROW_NUMBER() OVER (PARTITION BY source_id ORDER BY created_at DESC) = 1", + "sql_query": "SELECT source_id, facility_name, industry_sector, chimney_code, chimney_height_m, max_so2_limit_mg_nm3, max_nox_limit_mg_nm3, max_pm_limit_mg_nm3, region_zone, license_status, created_at FROM read_parquet('s3://dlh-bronze/staging/ref_emission_sources/*.parquet') QUALIFY ROW_NUMBER() OVER (PARTITION BY source_id ORDER BY created_at DESC) = 1", + "transformation_sql": "SELECT source_id, facility_name, industry_sector, chimney_code, chimney_height_m, max_so2_limit_mg_nm3, max_nox_limit_mg_nm3, max_pm_limit_mg_nm3, region_zone, license_status, created_at FROM read_parquet('s3://dlh-bronze/staging/ref_emission_sources/*.parquet') QUALIFY ROW_NUMBER() OVER (PARTITION BY source_id ORDER BY created_at DESC) = 1" + } + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-4059", + "type": "customNode", + "position": { + "x": 460, + "y": 80 + }, + "data": { + "label": "Extract Raw CEMS Telemetry", + "type": "source_db", + "subtitle": "Postgres / MySQL / Oracle", + "isSource": true, + "configured": true, + "config": { + "conn": "Postgres", + "conn_id": "conn_postgres_source_db", + "table": "public.raw_cems_telemetry_logs", + "table_name": "public.raw_cems_telemetry_logs", + "target_type": "minio_s3" + } + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-5075", + "type": "customNode", + "position": { + "x": 650, + "y": 80 + }, + "data": { + "label": "GE Telemetry Staging Check", + "type": "ge_quality", + "subtitle": "Great Expectations", + "configured": true, + "config": { + "ge_scope_level": "all", + "ge_table_name": "staging_cems_telemetry", + "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-6110", + "type": "customNode", + "position": { + "x": 840, + "y": 80 + }, + "data": { + "label": "DuckDB Telemetry to Bronze Iceberg", + "type": "duckdb_transform", + "subtitle": "In-process SQL (<512MB)", + "configured": true, + "config": { + "layer": "silver", + "source_parquet_path": "s3://dlh-bronze/staging/raw_cems_telemetry/*.parquet", + "source_path": "s3://dlh-bronze/staging/raw_cems_telemetry/*.parquet", + "target_table": "bronze_cems_telemetry_logs", + "table_name": "bronze_cems_telemetry_logs", + "primary_keys": "log_id", + "query": "SELECT log_id, source_id, telemetry_timestamp, so2_mg_nm3, nox_mg_nm3, pm_dust_mg_nm3, co_mg_nm3, flue_gas_flow_m3_s, flue_temp_celsius, sensor_status, created_at FROM read_parquet('s3://dlh-bronze/staging/raw_cems_telemetry/*.parquet') QUALIFY ROW_NUMBER() OVER (PARTITION BY log_id ORDER BY created_at DESC) = 1", + "sql_query": "SELECT log_id, source_id, telemetry_timestamp, so2_mg_nm3, nox_mg_nm3, pm_dust_mg_nm3, co_mg_nm3, flue_gas_flow_m3_s, flue_temp_celsius, sensor_status, created_at FROM read_parquet('s3://dlh-bronze/staging/raw_cems_telemetry/*.parquet') QUALIFY ROW_NUMBER() OVER (PARTITION BY log_id ORDER BY created_at DESC) = 1", + "transformation_sql": "SELECT log_id, source_id, telemetry_timestamp, so2_mg_nm3, nox_mg_nm3, pm_dust_mg_nm3, co_mg_nm3, flue_gas_flow_m3_s, flue_temp_celsius, sensor_status, created_at FROM read_parquet('s3://dlh-bronze/staging/raw_cems_telemetry/*.parquet') QUALIFY ROW_NUMBER() OVER (PARTITION BY log_id ORDER BY created_at DESC) = 1" + } + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-7127", + "type": "customNode", + "position": { + "x": 1030, + "y": 80 + }, + "data": { + "label": "GE Telemetry Bronze Quality", + "type": "ge_quality", + "subtitle": "Great Expectations", + "configured": true, + "config": { + "ge_scope_level": "all", + "ge_table_name": "bronze_cems_telemetry_logs", + "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-8143", + "type": "customNode", + "position": { + "x": 1220, + "y": 80 + }, + "data": { + "label": "dbt Silver CEMS Enriched", + "type": "dbt_transform", + "subtitle": "dbt-postgres", + "configured": true, + "config": { + "dbt_layer": "silver", + "dbt_model": "silver_cems_emission_enriched", + "sel": "silver_cems_emission_enriched", + "tgt": "postgres", + "trino_catalog": "iceberg", + "sql_query_preview": "-- dbt Silver Cleansed CEMS Telemetry Enriched with Master Sources\nSELECT \n t.log_id,\n t.telemetry_timestamp,\n t.source_id,\n s.facility_name,\n s.industry_sector,\n s.chimney_code,\n s.region_zone,\n t.so2_mg_nm3,\n s.max_so2_limit_mg_nm3,\n t.nox_mg_nm3,\n s.max_nox_limit_mg_nm3,\n t.pm_dust_mg_nm3,\n s.max_pm_limit_mg_nm3,\n t.co_mg_nm3,\n t.flue_gas_flow_m3_s,\n t.flue_temp_celsius,\n ROUND((t.so2_mg_nm3 * t.flue_gas_flow_m3_s * 3600.0) / 1000000.0, 3) AS so2_mass_rate_kg_h,\n ROUND((t.nox_mg_nm3 * t.flue_gas_flow_m3_s * 3600.0) / 1000000.0, 3) AS nox_mass_rate_kg_h,\n ROUND((t.pm_dust_mg_nm3 * t.flue_gas_flow_m3_s * 3600.0) / 1000000.0, 3) AS pm_mass_rate_kg_h,\n CASE \n WHEN t.so2_mg_nm3 > s.max_so2_limit_mg_nm3 OR t.nox_mg_nm3 > s.max_nox_limit_mg_nm3 OR t.pm_dust_mg_nm3 > s.max_pm_limit_mg_nm3 \n THEN 'EXCEEDING_LIMIT'\n ELSE 'COMPLIANT'\n END AS compliance_status,\n t.sensor_status\nFROM {{ ref('bronze_cems_telemetry_logs') }} t\nJOIN {{ ref('bronze_ref_emission_sources') }} s\n ON t.source_id = s.source_id\nWHERE t.sensor_status = 'VALID'", + "query": "-- dbt Silver Cleansed CEMS Telemetry Enriched with Master Sources\nSELECT \n t.log_id,\n t.telemetry_timestamp,\n t.source_id,\n s.facility_name,\n s.industry_sector,\n s.chimney_code,\n s.region_zone,\n t.so2_mg_nm3,\n s.max_so2_limit_mg_nm3,\n t.nox_mg_nm3,\n s.max_nox_limit_mg_nm3,\n t.pm_dust_mg_nm3,\n s.max_pm_limit_mg_nm3,\n t.co_mg_nm3,\n t.flue_gas_flow_m3_s,\n t.flue_temp_celsius,\n ROUND((t.so2_mg_nm3 * t.flue_gas_flow_m3_s * 3600.0) / 1000000.0, 3) AS so2_mass_rate_kg_h,\n ROUND((t.nox_mg_nm3 * t.flue_gas_flow_m3_s * 3600.0) / 1000000.0, 3) AS nox_mass_rate_kg_h,\n ROUND((t.pm_dust_mg_nm3 * t.flue_gas_flow_m3_s * 3600.0) / 1000000.0, 3) AS pm_mass_rate_kg_h,\n CASE \n WHEN t.so2_mg_nm3 > s.max_so2_limit_mg_nm3 OR t.nox_mg_nm3 > s.max_nox_limit_mg_nm3 OR t.pm_dust_mg_nm3 > s.max_pm_limit_mg_nm3 \n THEN 'EXCEEDING_LIMIT'\n ELSE 'COMPLIANT'\n END AS compliance_status,\n t.sensor_status\nFROM {{ ref('bronze_cems_telemetry_logs') }} t\nJOIN {{ ref('bronze_ref_emission_sources') }} s\n ON t.source_id = s.source_id\nWHERE t.sensor_status = 'VALID'" + } + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-9175", + "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_cems_emission_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-0193", + "type": "customNode", + "position": { + "x": 1600, + "y": 80 + }, + "data": { + "label": "dbt Gold Industrial Compliance Mart", + "type": "dbt_transform", + "subtitle": "dbt-postgres", + "configured": true, + "config": { + "dbt_layer": "gold", + "dbt_model": "gold_industry_emission_compliance_mart", + "sel": "gold_industry_emission_compliance_mart", + "tgt": "postgres", + "trino_catalog": "iceberg", + "sql_query_preview": "-- dbt Gold Industrial Emission Compliance & Load Mart\nSELECT \n region_zone,\n industry_sector,\n COUNT(log_id) AS total_telemetry_events,\n COUNT(DISTINCT source_id) AS active_chimneys_count,\n ROUND(AVG(so2_mg_nm3), 2) AS avg_so2_concentration,\n ROUND(AVG(nox_mg_nm3), 2) AS avg_nox_concentration,\n ROUND(AVG(pm_dust_mg_nm3), 2) AS avg_pm_concentration,\n ROUND(SUM(so2_mass_rate_kg_h) / 1000.0, 3) AS total_so2_emission_ton,\n ROUND(SUM(nox_mass_rate_kg_h) / 1000.0, 3) AS total_nox_emission_ton,\n ROUND(SUM(pm_mass_rate_kg_h) / 1000.0, 3) AS total_pm_emission_ton,\n COUNT(CASE WHEN compliance_status = 'EXCEEDING_LIMIT' THEN 1 END) AS total_violations_count,\n ROUND(\n (COUNT(CASE WHEN compliance_status = 'COMPLIANT' THEN 1 END)::NUMERIC / NULLIF(COUNT(log_id), 0)) * 100.0,\n 2\n ) AS compliance_rate_percent,\n CURRENT_TIMESTAMP AS mart_refreshed_at\nFROM {{ ref('silver_cems_emission_enriched') }}\nGROUP BY region_zone, industry_sector\nORDER BY total_violations_count DESC, total_so2_emission_ton DESC", + "query": "-- dbt Gold Industrial Emission Compliance & Load Mart\nSELECT \n region_zone,\n industry_sector,\n COUNT(log_id) AS total_telemetry_events,\n COUNT(DISTINCT source_id) AS active_chimneys_count,\n ROUND(AVG(so2_mg_nm3), 2) AS avg_so2_concentration,\n ROUND(AVG(nox_mg_nm3), 2) AS avg_nox_concentration,\n ROUND(AVG(pm_dust_mg_nm3), 2) AS avg_pm_concentration,\n ROUND(SUM(so2_mass_rate_kg_h) / 1000.0, 3) AS total_so2_emission_ton,\n ROUND(SUM(nox_mass_rate_kg_h) / 1000.0, 3) AS total_nox_emission_ton,\n ROUND(SUM(pm_mass_rate_kg_h) / 1000.0, 3) AS total_pm_emission_ton,\n COUNT(CASE WHEN compliance_status = 'EXCEEDING_LIMIT' THEN 1 END) AS total_violations_count,\n ROUND(\n (COUNT(CASE WHEN compliance_status = 'COMPLIANT' THEN 1 END)::NUMERIC / NULLIF(COUNT(log_id), 0)) * 100.0,\n 2\n ) AS compliance_rate_percent,\n CURRENT_TIMESTAMP AS mart_refreshed_at\nFROM {{ ref('silver_cems_emission_enriched') }}\nGROUP BY region_zone, industry_sector\nORDER BY total_violations_count DESC, total_so2_emission_ton DESC" + } + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-1241", + "type": "customNode", + "position": { + "x": 1790, + "y": 80 + }, + "data": { + "label": "GE Gold Compliance Mart Gate", + "type": "ge_quality", + "subtitle": "Great Expectations", + "configured": true, + "config": { + "ge_scope_level": "all", + "ge_table_name": "gold_industry_emission_compliance_mart", + "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-2409", + "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_industry_emission_compliance_mart", + "checksum": "sha256-verified-cems-emissions-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-1975", + "sourceHandle": null, + "target": "node-3042", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-1975-node-3042" + }, + { + "source": "node-3042", + "sourceHandle": null, + "target": "node-8143", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-3042-node-8143" + }, + { + "source": "node-4059", + "sourceHandle": null, + "target": "node-5075", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-4059-node-5075" + }, + { + "source": "node-5075", + "sourceHandle": null, + "target": "node-6110", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-5075-node-6110" + }, + { + "source": "node-6110", + "sourceHandle": null, + "target": "node-7127", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-6110-node-7127" + }, + { + "source": "node-7127", + "sourceHandle": null, + "target": "node-8143", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-7127-node-8143" + }, + { + "source": "node-8143", + "sourceHandle": null, + "target": "node-9175", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-8143-node-9175" + }, + { + "source": "node-9175", + "sourceHandle": null, + "target": "node-0193", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-9175-node-0193" + }, + { + "source": "node-0193", + "sourceHandle": null, + "target": "node-1241", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-0193-node-1241" + }, + { + "source": "node-1241", + "sourceHandle": null, + "target": "node-2409", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-1241-node-2409" + } + ], + "notes": [], + "is_draft": false, + "status": "deployed", + "updated_at": "2026-10-18T20:38:50.813467Z" +} \ No newline at end of file diff --git a/metadata/rbac_mappings.json b/metadata/rbac_mappings.json index d03fcc2..e748519 100644 --- a/metadata/rbac_mappings.json +++ b/metadata/rbac_mappings.json @@ -22,6 +22,6 @@ "admin_groups": [ "/Platform Administrators" ], - "updated_at": "2026-10-18T20:17:09.212620+00:00", + "updated_at": "2026-10-18T20:42:26.680284+00:00", "updated_by": "admin" } \ No newline at end of file