From ab5ba1d7a70cad07f13293c5b41e0d082ccd9585 Mon Sep 17 00:00:00 2001 From: syahril Date: Mon, 19 Oct 2026 04:04:02 +0700 Subject: [PATCH] feat(pipeline): deploy e2e_cems_industrial_emissions_flow from MERICA Canvas UI --- .../e2e_cems_industrial_emissions_flow.yaml | 76 ++++++++------- .../bronze/bronze_raw_cems_telemetry_logs.sql | 4 + .../bronze/bronze_ref_emission_sources.sql | 4 + .../e2e_cems_industrial_emissions_flow.json | 94 +++++++++---------- 4 files changed, 97 insertions(+), 81 deletions(-) create mode 100644 dbt/models/bronze/bronze_raw_cems_telemetry_logs.sql create mode 100644 dbt/models/bronze/bronze_ref_emission_sources.sql diff --git a/configs/e2e_cems_industrial_emissions_flow.yaml b/configs/e2e_cems_industrial_emissions_flow.yaml index 1885e85..353fbe0 100644 --- a/configs/e2e_cems_industrial_emissions_flow.yaml +++ b/configs/e2e_cems_industrial_emissions_flow.yaml @@ -8,7 +8,7 @@ governance: tier: Tier.Tier1 certification: Certification.Gold tasks: -- task_id: node-1975 +- task_id: node-6716 type: source_db label: Extract Master Emission Sources enabled: true @@ -19,12 +19,12 @@ tasks: table: public.ref_emission_sources table_name: public.ref_emission_sources target_type: minio_s3 -- task_id: node-3042 +- task_id: node-7780 type: duckdb_transform label: DuckDB Master to Bronze Iceberg enabled: true dependencies: - - node-1975 + - node-6716 config: layer: silver source_parquet_path: s3://dlh-bronze/staging/ref_emission_sources/*.parquet @@ -47,7 +47,7 @@ tasks: 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 +- task_id: node-8779 type: source_db label: Extract Raw CEMS Telemetry enabled: true @@ -58,12 +58,12 @@ tasks: table: public.raw_cems_telemetry_logs table_name: public.raw_cems_telemetry_logs target_type: minio_s3 -- task_id: node-5075 +- task_id: node-9796 type: ge_quality label: GE Telemetry Staging Check enabled: true dependencies: - - node-4059 + - node-8779 config: ge_scope_level: all ge_table_name: staging_cems_telemetry @@ -75,43 +75,45 @@ tasks: ge_col_table: stasiun_spku ge_target_column: stasiun_id ge_column_test_type: not_null - sql_assertion: '' - sql: '' + 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 +- task_id: node-0811 type: duckdb_transform label: DuckDB Telemetry to Bronze Iceberg enabled: true dependencies: - - node-5075 + - node-9796 config: layer: silver - source_parquet_path: s3://dlh-bronze/staging/raw_cems_telemetry/*.parquet - source_path: s3://dlh-bronze/staging/raw_cems_telemetry/*.parquet + source_parquet_path: s3://dlh-bronze/staging/raw_cems_telemetry_logs/*.parquet + source_path: s3://dlh-bronze/staging/raw_cems_telemetry_logs/*.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') + created_at FROM read_parquet('s3://dlh-bronze/staging/raw_cems_telemetry_logs/*.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') + created_at FROM read_parquet('s3://dlh-bronze/staging/raw_cems_telemetry_logs/*.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') + sensor_status, created_at FROM read_parquet('s3://dlh-bronze/staging/raw_cems_telemetry_logs/*.parquet') QUALIFY ROW_NUMBER() OVER (PARTITION BY log_id ORDER BY created_at DESC) = 1 -- task_id: node-7127 +- task_id: node-1818 type: ge_quality label: GE Telemetry Bronze Quality enabled: true dependencies: - - node-6110 + - node-0811 config: ge_scope_level: all ge_table_name: bronze_cems_telemetry_logs @@ -123,19 +125,21 @@ tasks: ge_col_table: stasiun_spku ge_target_column: stasiun_id ge_column_test_type: not_null - sql_assertion: '' - sql: '' + 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 +- task_id: node-2864 type: dbt_transform label: dbt Silver CEMS Enriched enabled: true dependencies: - - node-3042 - - node-7127 + - node-7780 + - node-1818 config: dbt_layer: silver dbt_model: silver_cems_emission_enriched @@ -170,12 +174,12 @@ tasks: \ 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 +- task_id: node-3898 type: ge_quality label: GE Silver Compliance Quality Gate enabled: true dependencies: - - node-8143 + - node-2864 config: ge_scope_level: all ge_table_name: silver_cems_emission_enriched @@ -187,18 +191,20 @@ tasks: ge_col_table: stasiun_spku ge_target_column: stasiun_id ge_column_test_type: not_null - sql_assertion: '' - sql: '' + 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 +- task_id: node-5020 type: dbt_transform label: dbt Gold Industrial Compliance Mart enabled: true dependencies: - - node-9175 + - node-3898 config: dbt_layer: gold dbt_model: gold_industry_emission_compliance_mart @@ -231,12 +237,12 @@ tasks: \ 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 +- task_id: node-6105 type: ge_quality label: GE Gold Compliance Mart Gate enabled: true dependencies: - - node-0193 + - node-5020 config: ge_scope_level: all ge_table_name: gold_industry_emission_compliance_mart @@ -248,18 +254,20 @@ tasks: ge_col_table: stasiun_spku ge_target_column: stasiun_id ge_column_test_type: not_null - sql_assertion: '' - sql: '' + 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 +- task_id: node-7152 type: sitomat_callback label: SITOMAT Governance Callback enabled: true dependencies: - - node-1241 + - node-6105 config: layer: gold artifact_ref: lakehouse.gold.gold_industry_emission_compliance_mart diff --git a/dbt/models/bronze/bronze_raw_cems_telemetry_logs.sql b/dbt/models/bronze/bronze_raw_cems_telemetry_logs.sql new file mode 100644 index 0000000..cc70b67 --- /dev/null +++ b/dbt/models/bronze/bronze_raw_cems_telemetry_logs.sql @@ -0,0 +1,4 @@ +-- dbt Bronze View Model: raw_cems_telemetry_logs +{{ config(materialized='view') }} + +SELECT * FROM {{ source('public', 'raw_cems_telemetry_logs') }} diff --git a/dbt/models/bronze/bronze_ref_emission_sources.sql b/dbt/models/bronze/bronze_ref_emission_sources.sql new file mode 100644 index 0000000..ed746c6 --- /dev/null +++ b/dbt/models/bronze/bronze_ref_emission_sources.sql @@ -0,0 +1,4 @@ +-- dbt Bronze View Model: ref_emission_sources +{{ config(materialized='view') }} + +SELECT * FROM {{ source('public', 'ref_emission_sources') }} diff --git a/metadata/e2e_cems_industrial_emissions_flow.json b/metadata/e2e_cems_industrial_emissions_flow.json index 001d87e..0a6a968 100644 --- a/metadata/e2e_cems_industrial_emissions_flow.json +++ b/metadata/e2e_cems_industrial_emissions_flow.json @@ -6,7 +6,7 @@ "owner_name": "Administrator Lokal (Break-Glass)", "nodes": [ { - "id": "node-1975", + "id": "node-6716", "type": "customNode", "position": { "x": 80, @@ -32,7 +32,7 @@ "dragging": false }, { - "id": "node-3042", + "id": "node-7780", "type": "customNode", "position": { "x": 270, @@ -61,7 +61,7 @@ "dragging": false }, { - "id": "node-4059", + "id": "node-8779", "type": "customNode", "position": { "x": 460, @@ -87,7 +87,7 @@ "dragging": false }, { - "id": "node-5075", + "id": "node-9796", "type": "customNode", "position": { "x": 650, @@ -123,7 +123,7 @@ "dragging": false }, { - "id": "node-6110", + "id": "node-0811", "type": "customNode", "position": { "x": 840, @@ -136,14 +136,14 @@ "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", + "source_parquet_path": "s3://dlh-bronze/staging/raw_cems_telemetry_logs/*.parquet", + "source_path": "s3://dlh-bronze/staging/raw_cems_telemetry_logs/*.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" + "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_logs/*.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_logs/*.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_logs/*.parquet') QUALIFY ROW_NUMBER() OVER (PARTITION BY log_id ORDER BY created_at DESC) = 1" } }, "width": 96, @@ -152,7 +152,7 @@ "dragging": false }, { - "id": "node-7127", + "id": "node-1818", "type": "customNode", "position": { "x": 1030, @@ -188,7 +188,7 @@ "dragging": false }, { - "id": "node-8143", + "id": "node-2864", "type": "customNode", "position": { "x": 1220, @@ -215,7 +215,7 @@ "dragging": false }, { - "id": "node-9175", + "id": "node-3898", "type": "customNode", "position": { "x": 1410, @@ -251,7 +251,7 @@ "dragging": false }, { - "id": "node-0193", + "id": "node-5020", "type": "customNode", "position": { "x": 1600, @@ -278,7 +278,7 @@ "dragging": false }, { - "id": "node-1241", + "id": "node-6105", "type": "customNode", "position": { "x": 1790, @@ -314,7 +314,7 @@ "dragging": false }, { - "id": "node-2409", + "id": "node-7152", "type": "customNode", "position": { "x": 1980, @@ -344,9 +344,9 @@ ], "edges": [ { - "source": "node-1975", + "source": "node-6716", "sourceHandle": null, - "target": "node-3042", + "target": "node-7780", "targetHandle": null, "type": "removableEdge", "style": { @@ -354,12 +354,12 @@ "strokeWidth": 2 }, "data": {}, - "id": "reactflow__edge-node-1975-node-3042" + "id": "reactflow__edge-node-6716-node-7780" }, { - "source": "node-3042", + "source": "node-7780", "sourceHandle": null, - "target": "node-8143", + "target": "node-2864", "targetHandle": null, "type": "removableEdge", "style": { @@ -367,12 +367,12 @@ "strokeWidth": 2 }, "data": {}, - "id": "reactflow__edge-node-3042-node-8143" + "id": "reactflow__edge-node-7780-node-2864" }, { - "source": "node-4059", + "source": "node-8779", "sourceHandle": null, - "target": "node-5075", + "target": "node-9796", "targetHandle": null, "type": "removableEdge", "style": { @@ -380,12 +380,12 @@ "strokeWidth": 2 }, "data": {}, - "id": "reactflow__edge-node-4059-node-5075" + "id": "reactflow__edge-node-8779-node-9796" }, { - "source": "node-5075", + "source": "node-9796", "sourceHandle": null, - "target": "node-6110", + "target": "node-0811", "targetHandle": null, "type": "removableEdge", "style": { @@ -393,12 +393,12 @@ "strokeWidth": 2 }, "data": {}, - "id": "reactflow__edge-node-5075-node-6110" + "id": "reactflow__edge-node-9796-node-0811" }, { - "source": "node-6110", + "source": "node-0811", "sourceHandle": null, - "target": "node-7127", + "target": "node-1818", "targetHandle": null, "type": "removableEdge", "style": { @@ -406,12 +406,12 @@ "strokeWidth": 2 }, "data": {}, - "id": "reactflow__edge-node-6110-node-7127" + "id": "reactflow__edge-node-0811-node-1818" }, { - "source": "node-7127", + "source": "node-1818", "sourceHandle": null, - "target": "node-8143", + "target": "node-2864", "targetHandle": null, "type": "removableEdge", "style": { @@ -419,12 +419,12 @@ "strokeWidth": 2 }, "data": {}, - "id": "reactflow__edge-node-7127-node-8143" + "id": "reactflow__edge-node-1818-node-2864" }, { - "source": "node-8143", + "source": "node-2864", "sourceHandle": null, - "target": "node-9175", + "target": "node-3898", "targetHandle": null, "type": "removableEdge", "style": { @@ -432,12 +432,12 @@ "strokeWidth": 2 }, "data": {}, - "id": "reactflow__edge-node-8143-node-9175" + "id": "reactflow__edge-node-2864-node-3898" }, { - "source": "node-9175", + "source": "node-3898", "sourceHandle": null, - "target": "node-0193", + "target": "node-5020", "targetHandle": null, "type": "removableEdge", "style": { @@ -445,12 +445,12 @@ "strokeWidth": 2 }, "data": {}, - "id": "reactflow__edge-node-9175-node-0193" + "id": "reactflow__edge-node-3898-node-5020" }, { - "source": "node-0193", + "source": "node-5020", "sourceHandle": null, - "target": "node-1241", + "target": "node-6105", "targetHandle": null, "type": "removableEdge", "style": { @@ -458,12 +458,12 @@ "strokeWidth": 2 }, "data": {}, - "id": "reactflow__edge-node-0193-node-1241" + "id": "reactflow__edge-node-5020-node-6105" }, { - "source": "node-1241", + "source": "node-6105", "sourceHandle": null, - "target": "node-2409", + "target": "node-7152", "targetHandle": null, "type": "removableEdge", "style": { @@ -471,11 +471,11 @@ "strokeWidth": 2 }, "data": {}, - "id": "reactflow__edge-node-1241-node-2409" + "id": "reactflow__edge-node-6105-node-7152" } ], "notes": [], "is_draft": false, "status": "deployed", - "updated_at": "2026-10-18T20:38:50.813467Z" + "updated_at": "2026-10-18T21:03:55.984345Z" } \ No newline at end of file