@ankit2604
Here is the combined, comprehensive architectural guide for building a Native Data Vault 2.0 Framework in Databricks (without `dbt`), incorporating modular Business Vault design, metadata-driven orchestration, incremental processing, dependency management, and data quality.
1. Core Architectural Strategy
A. Storage & Layout Optimizations
Delta Lake & Unity Catalog: Use Unity Catalog schemas (`rv` for Raw Vault, `bv` for Business Vault, `gold` for Information Marts).
Liquid Clustering (`CLUSTER BY`): Replaces legacy Z-Ordering and partitioning.
Hubs / Links: `ALTER TABLE rv.hub_customer CLUSTER BY (hk_customer_id);`
Satellites: `ALTER TABLE rv.sat_crm_customer CLUSTER BY (hk_customer_id, load_date);`
PIT Tables: `ALTER TABLE bv.pit_customer CLUSTER BY (snapshot_date, hk_customer_id);`
Bridge Tables: `ALTER TABLE bv.bridge_sales CLUSTER BY (snapshot_date, hk_order_id);`
Ghost Records: Insert a standard zero/ghost record (`-1` or `0x00000000000000000000000000000000`) into every Hub, Link, and Satellite during initialization to prevent `LEFT JOIN` record loss in downstream Marts.
2. Business Vault Implementation Patterns
A. Point-in-Time (PIT) Tables
PIT tables flatten non-equi joins across Satellites. Store only Hash Keys, Snapshot Dates, and Satellite `LOAD_DATE` values in PIT tables.
sql
CREATE OR REPLACE MATERIALIZED VIEW bv.pit_customer AS
SELECT
h.hk_customer_id,
sd.snapshot_date,
MAX(sat_crm.load_date) AS hk_sat_crm_customer_load_date,
MAX(sat_erp.load_date) AS hk_sat_erp_customer_load_date
FROM rv.hub_customer h
CROSS JOIN bv.dim_snapshot_date sd
LEFT JOIN rv.sat_crm_customer sat_crm
ON h.hk_customer_id = sat_crm.hk_customer_id
AND sat_crm.load_date <= sd.snapshot_date
LEFT JOIN rv.sat_erp_customer sat_erp
ON h.hk_customer_id = sat_erp.hk_customer_id
AND sat_erp.load_date <= sd.snapshot_date
GROUP BY h.hk_customer_id, sd.snapshot_date;
B. Bridge Tables & Derived Satellites
Bridge Tables: Flatten multi-hop Link joins (Customer $\rightarrow$ Order $\rightarrow$ Store) at snapshot dates into physical tables to accelerate Star Schema queries.
Derived / Computed Satellites: Compute business calculations, text standardization, and metric derivation cleanly separated from Raw Vault Satellites.
3. Incremental Processing Strategy
To maintain scalability without `dbt`, leverage Delta Lake primitives and Spark Structured Streaming:
A. Windowed Merge for PIT Tables
Avoid full PIT rebuilds by executing rolling window merges (e.g., last 30 days) using Delta `MERGE`:
sql
MERGE INTO bv.pit_customer AS target
USING (
SELECT
h.hk_customer_id,
sd.snapshot_date,
MAX(sat_crm.load_date) AS hk_sat_crm_customer_load_date,
MAX(sat_erp.load_date) AS hk_sat_erp_customer_load_date
FROM rv.hub_customer h
CROSS JOIN bv.dim_snapshot_date sd
LEFT JOIN rv.sat_crm_customer sat_crm
ON h.hk_customer_id = sat_crm.hk_customer_id
AND sat_crm.load_date <= sd.snapshot_date
LEFT JOIN rv.sat_erp_customer sat_erp
ON h.hk_customer_id = sat_erp.hk_customer_id
AND sat_erp.load_date <= sd.snapshot_date
WHERE sd.snapshot_date >= CURRENT_DATE() - INTERVAL 30 DAYS
GROUP BY h.hk_customer_id, sd.snapshot_date
) AS source
ON target.hk_customer_id = source.hk_customer_id
AND target.snapshot_date = source.snapshot_date
WHEN MATCHED THEN UPDATE SET
WHEN NOT MATCHED THEN INSERT ;
B. Delta Change Data Feed (CDF)
Enable `delta.enableChangeDataFeed = true` on Raw Satellites. Micro-batch pipelines can consume `readStream` from CDF to selectively recalculate updated PIT keys without scanning static data.
4. Native Dependency Management & Metadata Orchestration
Option 1: Declarative Lakeflow Pipelines (Recommended)
Databricks Lakeflow Pipelines automatically build the execution DAG based on table/view references:
python
import dlt
from pyspark.sql import functions as F
1. Hub
@dlt.table(name="hub_customer")
def hub_customer():
return spark.readStream.table("staging_customers").select("hk_customer_id", "customer_id", "load_date", "record_source")
2. Derived Satellite
@dlt.table(name="sat_customer_cleansed")
def sat_customer_cleansed():
return (
dlt.read_stream("sat_crm_customer")
.withColumn("standard_email", F.lower(F.trim(F.col("email"))))
)
3. PIT Table
@dlt.materialized_view(name="pit_customer")
def pit_customer():
return spark.sql("""
SELECT h.hk_customer_id, sd.snapshot_date, MAX(s.load_date) AS hk_sat_crm_load_date
FROM live.hub_customer h
CROSS JOIN live.dim_snapshot_date sd
LEFT JOIN live.sat_customer_cleansed s
ON h.hk_customer_id = s.hk_customer_id AND s.load_date <= sd.snapshot_date
GROUP BY h.hk_customer_id, sd.snapshot_date
""")
Option 2: Metadata-Driven Configuration Engine
If using standard Workflows, decouple transformation logic from orchestration using a metadata JSON/YAML schema driving PySpark generators:
json
{
"execution_tiers": [
{"tier": 1, "type": "Hubs_and_Links", "objects": ["hub_customer", "hub_order", "link_customer_order"]},
{"tier": 2, "type": "Raw_Satellites", "objects": ["sat_crm_customer", "sat_erp_order"]},
{"tier": 3, "type": "Business_Vault", "objects": ["pit_customer", "bridge_sales"]},
{"tier": 4, "type": "Information_Marts", "objects": ["dim_customer", "fact_orders"]}
]
}
5. Data Quality & Compliance Rules
Replaces external `dbt-expectations` with native Databricks enforcement features:
Lakeflow Expectations:
python
@dlt.table(name="sat_crm_customer")
@dlt.expect_or_fail("valid_hash_key", "hk_customer_id IS NOT NULL")
@dlt.expect_or_drop("valid_load_date", "load_date <= CURRENT_TIMESTAMP()")
@dlt.expect("valid_email_format", "email LIKE '%@%'")
def sat_crm_customer():
return spark.readStream.table("staging_crm_customers")
Unity Catalog Schema Constraints:
sql
ALTER TABLE rv.hub_customer ALTER COLUMN hk_customer_id SET NOT NULL;
ALTER TABLE rv.sat_crm_customer ADD CONSTRAINT valid_load_date CHECK (load_date > '2000-01-01');
Audit & Alerting: Query `system.access.audit` and Lakeflow event logs (`event_log(pipeline_id)`) to trigger SQL Alerts for orphan keys or non-conforming records.
