From abce291520df655996234fbdb4fd4e152e19907d Mon Sep 17 00:00:00 2001 From: syahril Date: Mon, 19 Oct 2026 03:21:06 +0700 Subject: [PATCH] feat(quality): emit Great Expectations test case results directly to OpenMetadata table test suites --- configs/e2e_medallion_permits_flow.yaml | 204 ++++++++ dags/helpers/ge_lakehouse_operator.py | 79 +++- metadata/e2e_medallion_permits_flow.json | 445 ++++++++++++++++++ ..._permits_medallion_dbt_postgres.draft.json | 304 ++++++++++++ ...gestion_spku_realtime_lakehouse.draft.json | 46 ++ metadata/rbac_mappings.json | 27 ++ 6 files changed, 1098 insertions(+), 7 deletions(-) create mode 100644 configs/e2e_medallion_permits_flow.yaml create mode 100644 metadata/e2e_medallion_permits_flow.json create mode 100644 metadata/example_permits_medallion_dbt_postgres.draft.json create mode 100644 metadata/ingestion_spku_realtime_lakehouse.draft.json create mode 100644 metadata/rbac_mappings.json diff --git a/configs/e2e_medallion_permits_flow.yaml b/configs/e2e_medallion_permits_flow.yaml new file mode 100644 index 0000000..f8ed7a9 --- /dev/null +++ b/configs/e2e_medallion_permits_flow.yaml @@ -0,0 +1,204 @@ +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 diff --git a/dags/helpers/ge_lakehouse_operator.py b/dags/helpers/ge_lakehouse_operator.py index 9c51ebb..61bd740 100644 --- a/dags/helpers/ge_lakehouse_operator.py +++ b/dags/helpers/ge_lakehouse_operator.py @@ -372,12 +372,78 @@ class GreatExpectationsLakehouseOperator(BaseLakehouseOperator): if not token: return - pipeline_svc = os.getenv("OPENMETADATA_PIPELINE_SERVICE_NAME", "dlh_airflow_service") - fqn = f"{pipeline_svc}.{self.dag_id or 'lakehouse_pipeline'}.{self.task_id}" + db_svc = os.getenv("OPENMETADATA_DB_SERVICE_NAME", "dlh_postgres_service") + db_name = os.getenv("POSTGRES_DB", "dlh_lakehouse") + raw_tbl = str(self.table_name or getattr(self, "ge_table_name", None) or "table") + parts = raw_tbl.split(".") + if len(parts) == 2: + schema = parts[0] + tbl_name = parts[1] + else: + tbl_name = parts[-1] + if tbl_name.startswith("gold_"): + schema = "gold" + elif tbl_name.startswith("silver_"): + schema = "silver" + elif tbl_name.startswith("bronze_"): + schema = "bronze" + elif "permit" in tbl_name.lower(): + schema = "bronze_lakehouse" + else: + schema = "bronze" + table_fqn = f"{db_svc}.{db_name}.{schema}.{tbl_name}" + + # 1. Ensure Table Test Suite + suite_payload = { + "name": f"{table_fqn}.testSuite", + "displayName": f"Quality Suite for {tbl_name}", + "basicEntityReference": table_fqn + } + suite_req = urllib.request.Request( + f"{om_host}/api/v1/dataQuality/testSuites/basic", + data=json.dumps(suite_payload).encode("utf-8"), + headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"}, + method="POST" + ) + try: + with urllib.request.urlopen(suite_req, timeout=3) as _: + pass + except Exception: + pass + + # 2. Ensure Test Case on Table EntityLink + tc_name = re.sub(r'[^a-zA-Z0-9_.-]', '_', f"{self.task_id}_quality").lower() + case_payload = { + "name": tc_name, + "displayName": f"GE Gate {self.task_id}", + "description": f"Great Expectations suite '{self.expectation_suite}' on {table_fqn}", + "entityLink": f"<#E::table::{table_fqn}>", + "testDefinition": "tableCustomSQLQuery", + "parameterValues": [ + {"name": "sqlExpression", "value": str(self.sql_assertion or getattr(self, "sql", "SELECT 1"))} + ] + } + tc_req = urllib.request.Request( + f"{om_host}/api/v1/dataQuality/testCases", + data=json.dumps(case_payload).encode("utf-8"), + headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"}, + method="POST" + ) + try: + with urllib.request.urlopen(tc_req, timeout=3) as _: + pass + except Exception: + pass + + # 3. Post Test Case Result to OpenMetadata Table Test Case + tc_fqn = f"{table_fqn}.{tc_name}" payload = { "timestamp": int(time.time() * 1000), "testCaseStatus": "Success" if summary["overall_success"] else "Failed", "result": f"Suite '{self.expectation_suite}': {summary['successful_evaluations']}/{summary['total_evaluations']} rules passed.", + "passedRows": summary.get("successful_evaluations", 1), + "failedRows": summary.get("total_unexpected_records", 0), + "passedRowsPercentage": float(summary.get("success_percent", 100.0)), "testResultValue": [ {"name": "evaluated_count", "value": str(summary["total_evaluations"])}, {"name": "unexpected_count", "value": str(summary["total_unexpected_records"])}, @@ -386,16 +452,15 @@ class GreatExpectationsLakehouseOperator(BaseLakehouseOperator): } req = urllib.request.Request( - f"{om_host}/api/v1/dataQuality/testCases/testCaseResult", + f"{om_host}/api/v1/dataQuality/testCases/testCaseResults/{urllib.parse.quote(tc_fqn, safe='')}", data=json.dumps(payload).encode("utf-8"), headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"}, - method="PUT" + method="POST" ) try: - with urllib.request.urlopen(req, timeout=3) as resp: - logger.info("✅ Data Quality metric emitted to OpenMetadata: %s", payload["testCaseStatus"]) + with urllib.request.urlopen(req, timeout=4) as resp: + logger.info("✅ Table Data Quality metric emitted to OpenMetadata: %s (%s)", tc_fqn, payload["testCaseStatus"]) except Exception: - # Fallback non-blocking pass except Exception as e: logger.debug("OpenMetadata DQ emission note: %s", e) diff --git a/metadata/e2e_medallion_permits_flow.json b/metadata/e2e_medallion_permits_flow.json new file mode 100644 index 0000000..71cba51 --- /dev/null +++ b/metadata/e2e_medallion_permits_flow.json @@ -0,0 +1,445 @@ +{ + "id": "e2e_medallion_permits_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-9377", + "type": "customNode", + "position": { + "x": 80, + "y": 80 + }, + "data": { + "label": "Source Database", + "type": "source_db", + "subtitle": "Postgres / MySQL / Oracle", + "isSource": true, + "configured": true, + "config": { + "conn": "Postgres", + "conn_id": "conn_postgres_source_db", + "table": "permits", + "table_name": "public.stasiun_spku", + "target_type": "minio_s3" + }, + "status": "warn", + "errorMessage": "Disconnected from workflow", + "missingFields": [], + "hasError": false, + "hasWarning": true + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-0405", + "type": "customNode", + "position": { + "x": 270, + "y": 80 + }, + "data": { + "label": "GE Quality Check", + "type": "ge_quality", + "subtitle": "Great Expectations", + "configured": true, + "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" + }, + "status": "warn", + "errorMessage": "Disconnected from workflow", + "missingFields": [], + "hasError": false, + "hasWarning": true + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-1406", + "type": "customNode", + "position": { + "x": 460, + "y": 80 + }, + "data": { + "label": "DuckDB Transform", + "type": "duckdb_transform", + "subtitle": "In-process SQL (<512MB)", + "configured": true, + "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" + }, + "status": "warn", + "errorMessage": "Disconnected from workflow", + "missingFields": [], + "hasError": false, + "hasWarning": true + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-2426", + "type": "customNode", + "position": { + "x": 650, + "y": 80 + }, + "data": { + "label": "GE Quality Check", + "type": "ge_quality", + "subtitle": "Great Expectations", + "configured": true, + "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" + }, + "status": "warn", + "errorMessage": "Disconnected from workflow", + "missingFields": [], + "hasError": false, + "hasWarning": true + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-3473", + "type": "customNode", + "position": { + "x": 840, + "y": 80 + }, + "data": { + "label": "dbt Transform", + "type": "dbt_transform", + "subtitle": "dbt-postgres", + "configured": true, + "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\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", + "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" + }, + "status": "warn", + "errorMessage": "Disconnected from workflow", + "missingFields": [], + "hasError": false, + "hasWarning": true + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-4489", + "type": "customNode", + "position": { + "x": 1030, + "y": 80 + }, + "data": { + "label": "GE Quality Check", + "type": "ge_quality", + "subtitle": "Great Expectations", + "configured": true, + "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" + }, + "status": "warn", + "errorMessage": "Disconnected from workflow", + "missingFields": [], + "hasError": false, + "hasWarning": true + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-5491", + "type": "customNode", + "position": { + "x": 1220, + "y": 80 + }, + "data": { + "label": "dbt Transform", + "type": "dbt_transform", + "subtitle": "dbt-postgres", + "configured": true, + "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" + }, + "status": "warn", + "errorMessage": "Disconnected from workflow", + "missingFields": [], + "hasError": false, + "hasWarning": true + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-6528", + "type": "customNode", + "position": { + "x": 1410, + "y": 80 + }, + "data": { + "label": "GE Quality Check", + "type": "ge_quality", + "subtitle": "Great Expectations", + "configured": true, + "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" + }, + "status": "warn", + "errorMessage": "Disconnected from workflow", + "missingFields": [], + "hasError": false, + "hasWarning": true + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-7512", + "type": "customNode", + "position": { + "x": 1600, + "y": 80 + }, + "data": { + "label": "Sitomat Callback", + "type": "sitomat_callback", + "subtitle": "Status callback", + "configured": true, + "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" + }, + "status": "warn", + "errorMessage": "Disconnected from workflow", + "missingFields": [], + "hasError": false, + "hasWarning": true + }, + "width": 96, + "height": 96, + "selected": true, + "dragging": false + } + ], + "edges": [ + { + "source": "node-9377", + "sourceHandle": null, + "target": "node-0405", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-9377-node-0405" + }, + { + "source": "node-0405", + "sourceHandle": null, + "target": "node-1406", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-0405-node-1406" + }, + { + "source": "node-1406", + "sourceHandle": null, + "target": "node-2426", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-1406-node-2426" + }, + { + "source": "node-2426", + "sourceHandle": null, + "target": "node-3473", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-2426-node-3473" + }, + { + "source": "node-3473", + "sourceHandle": null, + "target": "node-4489", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-3473-node-4489" + }, + { + "source": "node-4489", + "sourceHandle": null, + "target": "node-5491", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-4489-node-5491" + }, + { + "source": "node-5491", + "sourceHandle": null, + "target": "node-6528", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-5491-node-6528" + }, + { + "source": "node-6528", + "sourceHandle": null, + "target": "node-7512", + "targetHandle": null, + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {}, + "id": "reactflow__edge-node-6528-node-7512" + } + ], + "notes": [], + "is_draft": false, + "status": "deployed", + "updated_at": "2026-10-18T17:53:01.689174Z" +} \ No newline at end of file diff --git a/metadata/example_permits_medallion_dbt_postgres.draft.json b/metadata/example_permits_medallion_dbt_postgres.draft.json new file mode 100644 index 0000000..d7e7945 --- /dev/null +++ b/metadata/example_permits_medallion_dbt_postgres.draft.json @@ -0,0 +1,304 @@ +{ + "id": "example_permits_medallion_dbt_postgres", + "description": "Example Use Case: Environmental Permits Ingestion (Postgres DB) -> Bronze CDC Landing -> GE Quality Bronze -> dbt-postgres Silver -> GE Quality Silver -> dbt-postgres Gold Daily Mart", + "schedule": "0 8 * * *", + "owner_id": "local-admin", + "owner_name": "Administrator Lokal (Break-Glass)", + "nodes": [ + { + "id": "node-bronze-src", + "type": "customNode", + "position": { + "x": 60, + "y": 60 + }, + "data": { + "label": "Bronze DB Source (CDC Landing)", + "type": "source_db", + "configured": true, + "config": { + "source_type": "database", + "connection_id": "conn_postgres_source_db", + "table_name": "public.environmental_permits", + "target_table": "bronze_lakehouse.environmental_permits_cdc" + } + }, + "width": 96, + "height": 96 + }, + { + "id": "node-ge-bronze", + "type": "customNode", + "position": { + "x": 250, + "y": 60 + }, + "data": { + "label": "GE Quality Check (Bronze Suite)", + "type": "ge_quality", + "configured": true, + "config": { + "expectation_suite": "permits_bronze_suite", + "table_name": "bronze_lakehouse.environmental_permits_cdc", + "rules": [ + { + "type": "table_row_count_to_be_between", + "min_value": 1 + }, + { + "type": "column_values_to_not_be_null", + "column": "permit_id" + } + ], + "queries": [ + { + "name": "check_bronze_null_permit_id", + "sql": "SELECT count(*) FROM bronze_lakehouse.environmental_permits_cdc WHERE permit_id IS NULL", + "expected": 0, + "error_message": "Null permit_id detected in Bronze" + } + ] + } + }, + "width": 96, + "height": 96 + }, + { + "id": "node-sitomat-bronze", + "type": "customNode", + "position": { + "x": 440, + "y": 60 + }, + "data": { + "label": "Sitomat Bronze Callback", + "type": "sitomat_callback", + "configured": true, + "config": { + "layer": "bronze", + "artifact_ref": "dlh_lakehouse.bronze_lakehouse.environmental_permits_cdc", + "checksum": "auto-sha256" + } + }, + "width": 96, + "height": 96, + "selected": false, + "dragging": false + }, + { + "id": "node-silver-dbt", + "type": "customNode", + "position": { + "x": 630, + "y": 60 + }, + "data": { + "label": "dbt Silver Cleanse (dbt-postgres)", + "type": "dbt_transform", + "configured": true, + "config": { + "dbt_layer": "silver", + "dbt_model": "silver_environmental_permits_clean", + "target_table": "dlh_lakehouse.silver.silver_environmental_permits_clean" + } + }, + "width": 96, + "height": 96 + }, + { + "id": "node-ge-silver", + "type": "customNode", + "position": { + "x": 820, + "y": 60 + }, + "data": { + "label": "GE Quality Check (Silver Suite)", + "type": "ge_quality", + "configured": true, + "config": { + "expectation_suite": "permits_silver_suite", + "table_name": "silver.silver_environmental_permits_clean", + "rules": [ + { + "type": "column_values_to_not_be_null", + "column": "permit_id" + }, + { + "type": "column_values_to_be_in_set", + "column": "permit_status", + "allowed_set": [ + "APPROVED", + "IN_REVIEW", + "REJECTED" + ] + } + ], + "queries": [ + { + "name": "check_invalid_status_count", + "sql": "SELECT count(*) FROM silver.silver_environmental_permits_clean WHERE permit_status NOT IN ('APPROVED', 'IN_REVIEW', 'REJECTED')", + "expected": 0, + "error_message": "Invalid status detected in Silver" + } + ] + } + }, + "width": 96, + "height": 96 + }, + { + "id": "node-sitomat-silver", + "type": "customNode", + "position": { + "x": 1010, + "y": 60 + }, + "data": { + "label": "Sitomat Silver Callback", + "type": "sitomat_callback", + "configured": true, + "config": { + "layer": "silver", + "artifact_ref": "dlh_lakehouse.silver.silver_environmental_permits_clean", + "checksum": "auto-sha256" + } + }, + "width": 96, + "height": 96 + }, + { + "id": "node-gold-dbt", + "type": "customNode", + "position": { + "x": 1200, + "y": 60 + }, + "data": { + "label": "dbt Gold Mart (dbt-postgres)", + "type": "dbt_transform", + "configured": true, + "config": { + "dbt_layer": "gold", + "dbt_model": "gold_permits_summary_daily", + "target_table": "dlh_lakehouse.gold.gold_permits_summary_daily" + } + }, + "width": 96, + "height": 96 + }, + { + "id": "node-sitomat-gold", + "type": "customNode", + "position": { + "x": 1390, + "y": 60 + }, + "data": { + "label": "Sitomat Gold Callback (Data Product)", + "type": "sitomat_callback", + "configured": true, + "config": { + "layer": "gold", + "artifact_ref": "dlh_lakehouse.gold.gold_permits_summary_daily", + "checksum": "auto-sha256" + } + }, + "width": 96, + "height": 96 + } + ], + "edges": [ + { + "id": "e1-src-ge_bronze", + "source": "node-bronze-src", + "target": "node-ge-bronze", + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {} + }, + { + "id": "e2-ge_bronze-sitomat_bronze", + "source": "node-ge-bronze", + "target": "node-sitomat-bronze", + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {} + }, + { + "id": "e3-sitomat_bronze-silver", + "source": "node-sitomat-bronze", + "target": "node-silver-dbt", + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {} + }, + { + "id": "e4-silver-ge_silver", + "source": "node-silver-dbt", + "target": "node-ge-silver", + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {} + }, + { + "id": "e5-ge_silver-sitomat_silver", + "source": "node-ge-silver", + "target": "node-sitomat-silver", + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {} + }, + { + "id": "e6-sitomat_silver-gold", + "source": "node-sitomat-silver", + "target": "node-gold-dbt", + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {} + }, + { + "id": "e7-gold-sitomat_gold", + "source": "node-gold-dbt", + "target": "node-sitomat-gold", + "type": "removableEdge", + "style": { + "stroke": "var(--muted)", + "strokeWidth": 2 + }, + "data": {} + } + ], + "notes": [ + { + "id": "note-7937", + "text": "New note\nDouble-click to edit this note.", + "x": 190, + "y": -10, + "w": 300, + "h": 200, + "c": 0 + } + ], + "is_draft": true, + "status": "draft", + "updated_at": "2026-10-18T09:45:33.620665Z" +} \ No newline at end of file diff --git a/metadata/ingestion_spku_realtime_lakehouse.draft.json b/metadata/ingestion_spku_realtime_lakehouse.draft.json new file mode 100644 index 0000000..32bb582 --- /dev/null +++ b/metadata/ingestion_spku_realtime_lakehouse.draft.json @@ -0,0 +1,46 @@ +{ + "id": "ingestion_spku_realtime_lakehouse", + "description": "Dynamic Data Lakehouse Pipeline via DLH Studio", + "schedule": "0 */1 * * *", + "owner_id": "local-admin", + "owner_name": "Administrator Lokal (Break-Glass)", + "nodes": [ + { + "id": "node-7560", + "type": "customNode", + "position": { + "x": 80, + "y": 80 + }, + "data": { + "label": "dbt Transform", + "type": "dbt_transform", + "subtitle": "dbt-postgres", + "configured": true, + "config": { + "dbt_layer": "silver", + "dbt_model": "my_custom_model_silver", + "sel": "my_custom_model_silver", + "tgt": "postgres", + "trino_catalog": "iceberg", + "sql_query_preview": "SELECT stasiun_id, LOWER(nama_stasiun) AS nama_stasiun, ispu_val FROM {{ ref('bronze_spku_raw') }}", + "query": "SELECT stasiun_id, LOWER(nama_stasiun) AS nama_stasiun, ispu_val FROM {{ ref('bronze_spku_raw') }}" + }, + "status": "ok", + "errorMessage": "Configured & Valid", + "missingFields": [], + "hasError": false, + "hasWarning": false + }, + "width": 96, + "height": 96, + "selected": true, + "dragging": false + } + ], + "edges": [], + "notes": [], + "is_draft": true, + "status": "draft", + "updated_at": "2026-10-18T17:54:28.688466Z" +} \ No newline at end of file diff --git a/metadata/rbac_mappings.json b/metadata/rbac_mappings.json new file mode 100644 index 0000000..d03fcc2 --- /dev/null +++ b/metadata/rbac_mappings.json @@ -0,0 +1,27 @@ +{ + "editor_roles": [ + "data_engineer", + "pranata_komputer_ahli" + ], + "editor_groups": [ + "/Data Engineers", + "/DLH-Developers" + ], + "viewer_roles": [ + "merica_viewer", + "auditor" + ], + "viewer_groups": [ + "/Pimpinan Unit Kerja", + "/Auditor Internal BPKP" + ], + "admin_roles": [ + "merica_admin", + "superadmin" + ], + "admin_groups": [ + "/Platform Administrators" + ], + "updated_at": "2026-10-18T20:17:09.212620+00:00", + "updated_by": "admin" +} \ No newline at end of file