merica-dlh/merica-pipelines
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
b37ffc52ae0af7da035c97b21c0f0582e7dd9d0b
merica-pipelines/configs/e2e_cems_industrial_emissions_flow.yaml
T

280 lines
13 KiB
YAML
Raw Blame History

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
Reference in New Issue View Git Blame Copy Permalink