pipeline_id: e2e_modernized_ingestion_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-1048 type: source_db label: Source Database (Environmental Permits) enabled: true dependencies: [] config: conn: Postgres conn_id: conn_postgres_source_db table: public.environmental_permits table_name: public.environmental_permits target_type: minio_s3 watermark_column: updated_at ingestion_mode: incremental_watermark - task_id: node-2264 type: source_gws label: Source GWS (Waste Reports) enabled: true dependencies: [] config: url: '' spreadsheet_id: gws_sheet_1OSRugXZvfjoY9M0yySKT_sampah_2026 sheet_range: Sheet1 sheet_name: Sheet1 folder_id: 1OSRugXZvfjoY9M0yySKT drive_folder_id: 1OSRugXZvfjoY9M0yySKT selected_filename: Laporan_Sampah_Kabupaten_2026.xlsx header_row: 2 - task_id: node-3310 type: source_api_dlt label: Source REST API dlt (Sensor SPKU) enabled: true dependencies: [] config: endpoint_url: https://api.spku.dlh.id/v1/telemetry http_method: GET auth_type: none auth_token: '' pagination_strategy: page_number page_param: page limit_param: limit page_size: 100 cursor_column: measured_at lookback_hours: 3 unnest_mode: auto_flatten landing_bucket: dlh-landing landing_prefix: api/spku_hourly/ bronze_table: bronze_spku_telemetry url: https://api.spku.dlh.id/v1/telemetry - task_id: node-4363 type: duckdb_transform label: DuckDB Bronze Ingestion & Dedup enabled: true dependencies: - node-1048 - node-2264 - node-3310 config: layer: silver source_parquet_path: s3://dlh-bronze/staging/environmental_permits/*.parquet source_path: s3://dlh-bronze/staging/environmental_permits/*.parquet target_table: bronze_environmental_permits table_name: bronze_environmental_permits primary_keys: permit_id query: SELECT id AS permit_id, permit_number, company_name, permit_type, status, issued_date, updated_at FROM read_parquet('s3://dlh-bronze/staging/environmental_permits/*.parquet') QUALIFY ROW_NUMBER() OVER (PARTITION BY id ORDER BY updated_at DESC) = 1 sql_query: SELECT id AS permit_id, permit_number, company_name, permit_type, status, issued_date, updated_at FROM read_parquet('s3://dlh-bronze/staging/environmental_permits/*.parquet') QUALIFY ROW_NUMBER() OVER (PARTITION BY id ORDER BY updated_at DESC) = 1 transformation_sql: SELECT id AS permit_id, permit_number, company_name, permit_type, status, issued_date, updated_at FROM read_parquet('s3://dlh-bronze/staging/environmental_permits/*.parquet') QUALIFY ROW_NUMBER() OVER (PARTITION BY id ORDER BY updated_at DESC) = 1 - task_id: node-5408 type: dbt_transform label: dbt Silver Cleansed Permits Enriched enabled: true dependencies: - node-4363 config: dbt_layer: silver dbt_model: silver_permits_enriched sel: silver_permits_enriched tgt: postgres trino_catalog: iceberg sql_query_preview: "-- dbt Silver Cleansed Environmental Permits\nSELECT \n permit_id,\n\ \ permit_number,\n company_name,\n permit_type,\n status,\n issued_date,\n\ \ updated_at,\n CURRENT_TIMESTAMP AS dbt_validated_at\nFROM {{ ref('bronze_environmental_permits')\ \ }}\nWHERE status != 'REVISED'" query: "-- dbt Silver Cleansed Environmental Permits\nSELECT \n permit_id,\n\ \ permit_number,\n company_name,\n permit_type,\n status,\n issued_date,\n\ \ updated_at,\n CURRENT_TIMESTAMP AS dbt_validated_at\nFROM {{ ref('bronze_environmental_permits')\ \ }}\nWHERE status != 'REVISED'" test_configs: unique_columns: &id001 - permit_id - permit_number not_null_columns: &id002 - permit_id - status - updated_at accepted_values: {} enforce_contract: true unique_columns: *id001 not_null_columns: *id002 enforce_contract: true use_cosmos: true transformation_engine: cosmos - task_id: node-6444 type: ge_quality label: GE Silver Compliance Quality Gate enabled: true dependencies: - node-5408 config: ge_scope_level: table ge_table_name: silver_permits_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: '' sql: '' expected: 0 error_message: Unexpected rows detected in table suite: bronze fail: Stop DAG - task_id: node-7494 type: dbt_transform label: dbt Gold Regulatory Compliance Mart enabled: true dependencies: - node-6444 config: dbt_layer: gold dbt_model: gold_compliance_summary_mart sel: gold_compliance_summary_mart tgt: postgres trino_catalog: iceberg sql_query_preview: "-- dbt Gold Regulatory Compliance Mart\nSELECT \n permit_type,\n\ \ status,\n COUNT(permit_id) AS total_permits,\n COUNT(DISTINCT company_name)\ \ AS unique_companies,\n CURRENT_TIMESTAMP AS mart_refreshed_at\nFROM {{ ref('silver_permits_enriched')\ \ }}\nGROUP BY permit_type, status" query: "-- dbt Gold Regulatory Compliance Mart\nSELECT \n permit_type,\n status,\n\ \ COUNT(permit_id) AS total_permits,\n COUNT(DISTINCT company_name) AS unique_companies,\n\ \ CURRENT_TIMESTAMP AS mart_refreshed_at\nFROM {{ ref('silver_permits_enriched')\ \ }}\nGROUP BY permit_type, status" test_configs: unique_columns: - stasiun_id not_null_columns: - stasiun_id - ispu_val accepted_values: {} enforce_contract: false use_cosmos: true transformation_engine: cosmos - task_id: node-8543 type: sitomat_callback label: SITOMAT Governance Callback enabled: true dependencies: - node-7494 config: layer: gold artifact_ref: lakehouse.gold.gold_compliance_summary_mart checksum: sha256-verified-modernized-ingestion sitomat_base_url: http://localhost:3000 openmetadata_fqn: dlh_airflow_service.pipeline.task ge_status: PASSED url: /sitomat/callback stage: bronze