diff --git a/configs/e2e_modernized_ingestion_flow.yaml b/configs/e2e_modernized_ingestion_flow.yaml new file mode 100644 index 0000000..257b00d --- /dev/null +++ b/configs/e2e_modernized_ingestion_flow.yaml @@ -0,0 +1,185 @@ +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 diff --git a/dbt/models/bronze/bronze_environmental_permits.sql b/dbt/models/bronze/bronze_environmental_permits.sql new file mode 100644 index 0000000..995424c --- /dev/null +++ b/dbt/models/bronze/bronze_environmental_permits.sql @@ -0,0 +1,4 @@ +-- dbt Bronze View Model: environmental_permits +{{ config(materialized='view') }} + +SELECT * FROM {{ source('public', 'environmental_permits') }} diff --git a/dbt/models/gold/gold_compliance_summary_mart.sql b/dbt/models/gold/gold_compliance_summary_mart.sql new file mode 100644 index 0000000..8336f8b --- /dev/null +++ b/dbt/models/gold/gold_compliance_summary_mart.sql @@ -0,0 +1,9 @@ +-- dbt Gold Regulatory Compliance Mart +SELECT + permit_type, + status, + COUNT(permit_id) AS total_permits, + COUNT(DISTINCT company_name) AS unique_companies, + CURRENT_TIMESTAMP AS mart_refreshed_at +FROM {{ ref('silver_permits_enriched') }} +GROUP BY permit_type, status \ No newline at end of file diff --git a/dbt/models/gold/schema.yml b/dbt/models/gold/schema.yml new file mode 100644 index 0000000..3fc2d9e --- /dev/null +++ b/dbt/models/gold/schema.yml @@ -0,0 +1,13 @@ +version: 2 +models: +- name: gold_compliance_summary_mart + description: dbt Model gold_compliance_summary_mart auto-registered with Shift-Left + quality assertions + columns: + - name: ispu_val + tests: + - not_null + - name: stasiun_id + tests: + - unique + - not_null diff --git a/dbt/models/silver/schema.yml b/dbt/models/silver/schema.yml new file mode 100644 index 0000000..d5f1489 --- /dev/null +++ b/dbt/models/silver/schema.yml @@ -0,0 +1,22 @@ +version: 2 +models: +- name: silver_permits_enriched + description: dbt Model silver_permits_enriched auto-registered with Shift-Left quality + assertions + config: + contract: + enforced: true + columns: + - name: permit_id + tests: + - unique + - not_null + - name: permit_number + tests: + - unique + - name: status + tests: + - not_null + - name: updated_at + tests: + - not_null diff --git a/dbt/models/silver/silver_permits_enriched.sql b/dbt/models/silver/silver_permits_enriched.sql new file mode 100644 index 0000000..f2a32d3 --- /dev/null +++ b/dbt/models/silver/silver_permits_enriched.sql @@ -0,0 +1,12 @@ +-- dbt Silver Cleansed Environmental Permits +SELECT + permit_id, + permit_number, + company_name, + permit_type, + status, + issued_date, + updated_at, + CURRENT_TIMESTAMP AS dbt_validated_at +FROM {{ ref('bronze_environmental_permits') }} +WHERE status != 'REVISED' \ No newline at end of file diff --git a/dbt/models/sources.yml b/dbt/models/sources.yml index 92e5023..49663a2 100644 --- a/dbt/models/sources.yml +++ b/dbt/models/sources.yml @@ -31,3 +31,6 @@ sources: pipeline - name: stasiun_spku description: Auto-registered source table 'stasiun_spku' from canvas pipeline + - name: environmental_permits + description: Auto-registered source table 'environmental_permits' from canvas + pipeline diff --git a/metadata/e2e_modernized_ingestion_flow.json b/metadata/e2e_modernized_ingestion_flow.json new file mode 100644 index 0000000..58a1d4a --- /dev/null +++ b/metadata/e2e_modernized_ingestion_flow.json @@ -0,0 +1,384 @@ +{ + "id": "e2e_modernized_ingestion_flow", + "description": "Dynamic Data Lakehouse Pipeline via DLH Studio", + "schedule": "0 */1 * * *", + "owner_id": "local-admin", + "owner_name": "Administrator Lokal (Break-Glass)", + "nodes": [ + { + "id": "node-1048", + "type": "customNode", + "position": { + "x": 80, + "y": 80 + }, + "data": { + "label": "Source Database (Environmental Permits)", + "type": "source_db", + "subtitle": "Postgres / MySQL / Oracle", + "isSource": true, + "configured": true, + "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" + } + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-2264", + "type": "customNode", + "position": { + "x": 270, + "y": 80 + }, + "data": { + "label": "Source GWS (Waste Reports)", + "type": "source_gws", + "subtitle": "Sheets & Drive PDF", + "isSource": true, + "configured": true, + "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 + } + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-3310", + "type": "customNode", + "position": { + "x": 460, + "y": 80 + }, + "data": { + "label": "Source REST API dlt (Sensor SPKU)", + "type": "source_api_dlt", + "subtitle": "Auto Un-nesting & Pagination", + "isSource": true, + "configured": true, + "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" + } + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-4363", + "type": "customNode", + "position": { + "x": 650, + "y": 80 + }, + "data": { + "label": "DuckDB Bronze Ingestion & Dedup", + "type": "duckdb_transform", + "subtitle": "In-process SQL (<512MB)", + "configured": true, + "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" + } + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-5408", + "type": "customNode", + "position": { + "x": 840, + "y": 80 + }, + "data": { + "label": "dbt Silver Cleansed Permits Enriched", + "type": "dbt_transform", + "subtitle": "dbt-postgres", + "configured": true, + "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": [ + "permit_id", + "permit_number" + ], + "not_null_columns": [ + "permit_id", + "status", + "updated_at" + ], + "accepted_values": {}, + "enforce_contract": true + }, + "unique_columns": [ + "permit_id", + "permit_number" + ], + "not_null_columns": [ + "permit_id", + "status", + "updated_at" + ], + "enforce_contract": true, + "use_cosmos": true, + "transformation_engine": "cosmos" + } + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-6444", + "type": "customNode", + "position": { + "x": 1030, + "y": 80 + }, + "data": { + "label": "GE Silver Compliance Quality Gate", + "type": "ge_quality", + "subtitle": "Great Expectations", + "configured": true, + "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" + } + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-7494", + "type": "customNode", + "position": { + "x": 1220, + "y": 80 + }, + "data": { + "label": "dbt Gold Regulatory Compliance Mart", + "type": "dbt_transform", + "subtitle": "dbt-postgres", + "configured": true, + "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" + } + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-8543", + "type": "customNode", + "position": { + "x": 1410, + "y": 80 + }, + "data": { + "label": "SITOMAT Governance Callback", + "type": "sitomat_callback", + "subtitle": "Status callback", + "configured": true, + "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" + } + }, + "width": 96, + "height": 96, + "selected": true, + "dragging": false + } + ], + "edges": [ + { + "source": "node-1048", + "sourceHandle": null, + "target": "node-4363", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-1048-node-4363" + }, + { + "source": "node-2264", + "sourceHandle": null, + "target": "node-4363", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-2264-node-4363" + }, + { + "source": "node-3310", + "sourceHandle": null, + "target": "node-4363", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-3310-node-4363" + }, + { + "source": "node-4363", + "sourceHandle": null, + "target": "node-5408", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-4363-node-5408" + }, + { + "source": "node-5408", + "sourceHandle": null, + "target": "node-6444", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-5408-node-6444" + }, + { + "source": "node-6444", + "sourceHandle": null, + "target": "node-7494", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-6444-node-7494" + }, + { + "source": "node-7494", + "sourceHandle": null, + "target": "node-8543", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-7494-node-8543" + } + ], + "notes": [], + "is_draft": false, + "status": "deployed", + "updated_at": "2026-10-19T09:42:04.272325Z" +} \ No newline at end of file diff --git a/metadata/rbac_mappings.json b/metadata/rbac_mappings.json index e43db0c..ab08a24 100644 --- a/metadata/rbac_mappings.json +++ b/metadata/rbac_mappings.json @@ -22,6 +22,6 @@ "admin_groups": [ "/Platform Administrators" ], - "updated_at": "2026-10-19T09:01:42.813803+00:00", + "updated_at": "2026-10-19T09:46:41.835103+00:00", "updated_by": "admin" } \ No newline at end of file