merica-pipelines/configs/e2e_modernized_ingestion_flow.yaml

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-7278
  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-8718
  type: source_gws
  label: Source GWS (Waste Reports)
  enabled: true
  dependencies: []
  config:
    url: gws_sheet_1OSRugXZvfjoY9M0yySKT_sampah_2026
    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-0372
  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-2418
  type: duckdb_transform
  label: DuckDB Bronze Ingestion & Dedup
  enabled: true
  dependencies:
  - node-7278
  - node-8718
  - node-0372
  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 permit_id, permit_number, company_name, activity_type AS permit_type,
      status, issued_date, updated_at FROM read_parquet('s3://dlh-bronze/staging/environmental_permits/*.parquet')
      QUALIFY ROW_NUMBER() OVER (PARTITION BY permit_id ORDER BY updated_at DESC)
      = 1
    sql_query: SELECT permit_id, permit_number, company_name, activity_type AS permit_type,
      status, issued_date, updated_at FROM read_parquet('s3://dlh-bronze/staging/environmental_permits/*.parquet')
      QUALIFY ROW_NUMBER() OVER (PARTITION BY permit_id ORDER BY updated_at DESC)
      = 1
    transformation_sql: SELECT permit_id, permit_number, company_name, activity_type
      AS permit_type, status, issued_date, updated_at FROM read_parquet('s3://dlh-bronze/staging/environmental_permits/*.parquet')
      QUALIFY ROW_NUMBER() OVER (PARTITION BY permit_id ORDER BY updated_at DESC)
      = 1
- task_id: node-4451
  type: dbt_transform
  label: dbt Silver Cleansed Permits Enriched
  enabled: true
  dependencies:
  - node-2418
  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-6068
  type: ge_quality
  label: GE Silver Compliance Quality Gate
  enabled: true
  dependencies:
  - node-4451
  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-8030
  type: dbt_transform
  label: dbt Gold Regulatory Compliance Mart
  enabled: true
  dependencies:
  - node-6068
  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-9428
  type: sitomat_callback
  label: SITOMAT Governance Callback
  enabled: true
  dependencies:
  - node-8030
  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