190 lines
6.4 KiB
YAML
190 lines
6.4 KiB
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-5630
|
||
|
|
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-6855
|
||
|
|
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-8156
|
||
|
|
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-9404
|
||
|
|
type: duckdb_transform
|
||
|
|
label: DuckDB Bronze Ingestion & Dedup
|
||
|
|
enabled: true
|
||
|
|
dependencies:
|
||
|
|
- node-5630
|
||
|
|
- node-6855
|
||
|
|
- node-8156
|
||
|
|
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-0708
|
||
|
|
type: dbt_transform
|
||
|
|
label: dbt Silver Cleansed Permits Enriched
|
||
|
|
enabled: true
|
||
|
|
dependencies:
|
||
|
|
- node-9404
|
||
|
|
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-2234
|
||
|
|
type: ge_quality
|
||
|
|
label: GE Silver Compliance Quality Gate
|
||
|
|
enabled: true
|
||
|
|
dependencies:
|
||
|
|
- node-0708
|
||
|
|
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-3694
|
||
|
|
type: dbt_transform
|
||
|
|
label: dbt Gold Regulatory Compliance Mart
|
||
|
|
enabled: true
|
||
|
|
dependencies:
|
||
|
|
- node-2234
|
||
|
|
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: &id003 []
|
||
|
|
not_null_columns: &id004
|
||
|
|
- permit_type
|
||
|
|
- status
|
||
|
|
accepted_values: {}
|
||
|
|
enforce_contract: false
|
||
|
|
unique_columns: *id003
|
||
|
|
not_null_columns: *id004
|
||
|
|
use_cosmos: true
|
||
|
|
transformation_engine: cosmos
|
||
|
|
- task_id: node-4977
|
||
|
|
type: sitomat_callback
|
||
|
|
label: SITOMAT Governance Callback
|
||
|
|
enabled: true
|
||
|
|
dependencies:
|
||
|
|
- node-3694
|
||
|
|
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
|