205 lines
7.5 KiB
YAML
205 lines
7.5 KiB
YAML
|
pipeline_id: e2e_medallion_permits_flow
|
||
|
|
description: Dynamic Data Lakehouse Pipeline via DLH Studio
|
||
|
|
schedule: 0 */1 * * *
|
||
|
|
mode: flexible
|
||
|
|
tasks:
|
||
|
|
- task_id: node-9377
|
||
|
|
type: source_db
|
||
|
|
label: Source Database
|
||
|
|
enabled: true
|
||
|
|
dependencies: []
|
||
|
|
config:
|
||
|
|
conn: Postgres
|
||
|
|
conn_id: conn_postgres_source_db
|
||
|
|
table: permits
|
||
|
|
table_name: public.stasiun_spku
|
||
|
|
target_type: minio_s3
|
||
|
|
- task_id: node-0405
|
||
|
|
type: ge_quality
|
||
|
|
label: GE Quality Check
|
||
|
|
enabled: true
|
||
|
|
dependencies:
|
||
|
|
- node-9377
|
||
|
|
config:
|
||
|
|
ge_scope_level: all
|
||
|
|
ge_table_name: staging_environmental_permits
|
||
|
|
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-1406
|
||
|
|
type: duckdb_transform
|
||
|
|
label: DuckDB Transform
|
||
|
|
enabled: true
|
||
|
|
dependencies:
|
||
|
|
- node-0405
|
||
|
|
config:
|
||
|
|
layer: silver
|
||
|
|
source_parquet_path: s3://dlh-bronze/db_cdc/environmental_permits/
|
||
|
|
source_path: s3://dlh-bronze/db_cdc/environmental_permits/
|
||
|
|
target_table: dlh_lakehouse.silver.environmental_permits
|
||
|
|
table_name: dlh_lakehouse.silver.environmental_permits
|
||
|
|
primary_keys: id, updated_at
|
||
|
|
query: SELECT * FROM source_tbl
|
||
|
|
sql_query: SELECT * FROM source_parquet
|
||
|
|
transformation_sql: SELECT permit_id, permit_number, company_name, business_type,
|
||
|
|
location, status, environmental_score, created_at, 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-2426
|
||
|
|
type: ge_quality
|
||
|
|
label: GE Quality Check
|
||
|
|
enabled: true
|
||
|
|
dependencies:
|
||
|
|
- node-1406
|
||
|
|
config:
|
||
|
|
ge_scope_level: all
|
||
|
|
ge_table_name: bronze_environmental_permits
|
||
|
|
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-3473
|
||
|
|
type: dbt_transform
|
||
|
|
label: dbt Transform
|
||
|
|
enabled: true
|
||
|
|
dependencies:
|
||
|
|
- node-2426
|
||
|
|
config:
|
||
|
|
dbt_layer: silver
|
||
|
|
dbt_model: stg_spku_realtime
|
||
|
|
sel: silver
|
||
|
|
tgt: postgres
|
||
|
|
trino_catalog: iceberg
|
||
|
|
sql_query_preview: "-- dbt Silver Cleansed Model with Master Table Enrichment\n\
|
||
|
|
SELECT \n p.permit_id,\n p.permit_number,\n p.company_name,\n p.business_type,\n\
|
||
|
|
\ m.category_name,\n m.risk_level,\n m.compliance_weight,\n p.environmental_score,\n\
|
||
|
|
\ ROUND(p.environmental_score * m.compliance_weight, 2) AS weighted_compliance_score,\n\
|
||
|
|
\ p.status,\n p.location,\n p.updated_at\nFROM {{ ref('bronze_environmental_permits')\
|
||
|
|
\ }} p\nLEFT JOIN {{ ref('ref_business_categories') }} m \n ON p.business_type\
|
||
|
|
\ = m.business_type"
|
||
|
|
query: "-- dbt Silver Cleansed Model with Master Table Enrichment\nSELECT \n \
|
||
|
|
\ p.permit_id,\n p.permit_number,\n p.company_name,\n p.business_type,\n\
|
||
|
|
\ m.category_name,\n m.risk_level,\n m.compliance_weight,\n p.environmental_score,\n\
|
||
|
|
\ ROUND(p.environmental_score * m.compliance_weight, 2) AS weighted_compliance_score,\n\
|
||
|
|
\ p.status,\n p.location,\n p.updated_at\nFROM {{ ref('bronze_environmental_permits')\
|
||
|
|
\ }} p\nLEFT JOIN {{ ref('ref_business_categories') }} m \n ON p.business_type\
|
||
|
|
\ = m.business_type"
|
||
|
|
- task_id: node-4489
|
||
|
|
type: ge_quality
|
||
|
|
label: GE Quality Check
|
||
|
|
enabled: true
|
||
|
|
dependencies:
|
||
|
|
- node-3473
|
||
|
|
config:
|
||
|
|
ge_scope_level: all
|
||
|
|
ge_table_name: silver_environmental_compliance
|
||
|
|
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-5491
|
||
|
|
type: dbt_transform
|
||
|
|
label: dbt Transform
|
||
|
|
enabled: true
|
||
|
|
dependencies:
|
||
|
|
- node-4489
|
||
|
|
config:
|
||
|
|
dbt_layer: silver
|
||
|
|
dbt_model: stg_spku_realtime
|
||
|
|
sel: silver
|
||
|
|
tgt: postgres
|
||
|
|
trino_catalog: iceberg
|
||
|
|
sql_query_preview: "-- dbt Gold Aggregated Risk Mart\nSELECT \n risk_level,\n\
|
||
|
|
\ business_type,\n category_name,\n COUNT(permit_id) AS total_permits,\n\
|
||
|
|
\ COUNT(CASE WHEN status = 'ACTIVE' THEN 1 END) AS active_permits,\n COUNT(CASE\
|
||
|
|
\ WHEN status = 'SUSPENDED' THEN 1 END) AS suspended_permits,\n ROUND(AVG(environmental_score),\
|
||
|
|
\ 2) AS avg_environmental_score,\n ROUND(AVG(weighted_compliance_score), 2)\
|
||
|
|
\ AS avg_weighted_score,\n CURRENT_TIMESTAMP AS calculated_at\nFROM {{ ref('silver_environmental_compliance')\
|
||
|
|
\ }}\nGROUP BY risk_level, business_type, category_name\nORDER BY avg_weighted_score\
|
||
|
|
\ DESC"
|
||
|
|
query: "-- dbt Gold Aggregated Risk Mart\nSELECT \n risk_level,\n business_type,\n\
|
||
|
|
\ category_name,\n COUNT(permit_id) AS total_permits,\n COUNT(CASE WHEN status\
|
||
|
|
\ = 'ACTIVE' THEN 1 END) AS active_permits,\n COUNT(CASE WHEN status = 'SUSPENDED'\
|
||
|
|
\ THEN 1 END) AS suspended_permits,\n ROUND(AVG(environmental_score), 2) AS\
|
||
|
|
\ avg_environmental_score,\n ROUND(AVG(weighted_compliance_score), 2) AS avg_weighted_score,\n\
|
||
|
|
\ CURRENT_TIMESTAMP AS calculated_at\nFROM {{ ref('silver_environmental_compliance')\
|
||
|
|
\ }}\nGROUP BY risk_level, business_type, category_name\nORDER BY avg_weighted_score\
|
||
|
|
\ DESC"
|
||
|
|
- task_id: node-6528
|
||
|
|
type: ge_quality
|
||
|
|
label: GE Quality Check
|
||
|
|
enabled: true
|
||
|
|
dependencies:
|
||
|
|
- node-5491
|
||
|
|
config:
|
||
|
|
ge_scope_level: all
|
||
|
|
ge_table_name: gold_environmental_risk_summary
|
||
|
|
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-7512
|
||
|
|
type: sitomat_callback
|
||
|
|
label: Sitomat Callback
|
||
|
|
enabled: true
|
||
|
|
dependencies:
|
||
|
|
- node-6528
|
||
|
|
config:
|
||
|
|
layer: bronze
|
||
|
|
artifact_ref: lakehouse.bronze.dataset_cdc
|
||
|
|
checksum: sha256-verified-pipeline-pass
|
||
|
|
sitomat_base_url: http://localhost:3000
|
||
|
|
openmetadata_fqn: dlh_airflow_service.pipeline.task
|
||
|
|
ge_status: PASSED
|
||
|
|
url: /sitomat/callback
|
||
|
|
stage: bronze
|