Databricks CDC Lakehouse Lab

End-to-end reference implementation of a Change Data Capture pipeline from the dvdrental PostgreSQL sample database into a Databricks medallion lakehouse (Bronze → Silver → Vault → Gold). It also includes a second, independent near-real-time pipeline that ingests live trading events from pump.fun, the Solana meme-coin launchpad.

Agent-system inspiration was borrowed from agency-agents.

Architecture

PostgreSQL dvdrental (WAL)
   → Debezium Connect (topics: cdc.public.* — all 15 tables)
     → Databricks Bronze (raw Debezium envelopes in Delta tables)
       → Databricks Silver (current-state via MERGE, schema evolution)
         → Databricks Vault (Data Vault 2.0: Hubs / Links / Satellites / PIT / Bridge)
           → dbt Gold (business-ready models with data quality tests)

pump.fun (Solana on-chain trades)
   → PumpAPI websocket → ingestion/pumpfun service (outside Databricks)
     → batch → JSONL → zstd-compress → base64-encode → Unity Catalog Volume
       → pumpapi-lakehouse (Lakeflow Declarative Pipeline: decode → decompress → parse)
         → Bronze (pump_events_raw) + 3 Silver tables (events / tokens / transfers)

See docs/architecture.md for the full pump.fun pipeline design.

Directory Structure

pumpapi-lakehouse/           # pump.fun Bronze + Silver — one Lakeflow Declarative Pipeline
└── transformations/
    ├── bronze_pump_events.py      ← decode zstd+base64 → workspace.bronze.pump_events_raw
    ├── silver_pump_events.py      ← all events, typed  → silver.pump_events
    ├── silver_pump_tokens.py      ← action=create      → silver.pump_tokens
    └── silver_pump_transfers.py   ← action=transfer     → silver.pump_transfers

ingestion/                  # Ingestion layer (Bronze)
├── consumers/
│   ├── NB_ingest_to_bronze.ipynb                    ← dvdrental Bronze streaming notebook
│   ├── outdated__NB_ingest_pumpfun_to_bronze.ipynb  ← superseded, kept for reference
│   └── outdated__NB_dlt_pumpfun_bronze.ipynb        ← superseded, kept for reference
├── cdc/
│   └── postgres-connector.json     ← Debezium connector config
├── pumpfun/                 # Standalone pump.fun websocket ingestor (runs outside Databricks)
│   ├── app/                        ← websocket client → zstd+base64 batch writer → Databricks uploader
│   ├── Dockerfile / docker-compose.yml
│   ├── systemd/pumpfun-ingestor.service
│   └── README.md
└── generators/             # Data mutation scripts
    ├── load_generator.py           ← Rental/payment generator
    ├── load_products_generator.py  ← Film update generator
    └── load_bulk_data.py           ← Bulk data seeder

processing/                 # All processing logic
├── silver/
│   ├── NB_process_to_silver_generic.ipynb        ← Metadata-driven Bronze → Silver (dvdrental)
│   └── outdated__NB_process_pumpfun_silver.ipynb ← superseded, kept for reference
├── vault/
│   ├── NB_ingest_to_hubs.ipynb
│   ├── NB_ingest_to_links.ipynb
│   ├── NB_ingest_to_satellites.ipynb
│   └── NB_dv_business_vault.ipynb
└── common/                 # Shared helper notebooks
    ├── NB_silver_metadata.ipynb
    ├── NB_catalog_helpers.ipynb
    ├── NB_schema_contracts.ipynb
    └── NB_schema_drift_helpers.ipynb

transformation/             # dbt (vault + gold)
├── NB_run_dbt.ipynb        ← Databricks notebook that runs dbt (dbtRunner API)
└── dbt_project/            ← dbt project
    ├── dbt_project.yml
    └── models/
        ├── vault/          ← Incremental vault models (hubs/links/sats/pit/bridge)
        └── gold/           ← Gold data marts

config/                     # All configuration files
├── datavault/
│   └── dv_model.json       ← Vault config (uploaded to Unity Catalog Volume)
├── silver/
│   └── configs/dvdrental/  ← Per-table Silver configs (15 JSON files)
└── .envexample

orchestration/              # Jobs / workflows
└── bundle/                 ← Databricks Asset Bundles (future)

scripts/                    # CLI utilities
├── deploy_jobs.py          ← Deploy all Databricks jobs (dvdrental + pumpfun)
├── push_secrets_to_databricks.py
├── upload_vault_config.py
└── reset_checkpoints.py

infra/                      # Local environment setup
└── docker/
    ├── docker-compose.yml
    └── init-dvdrental.sh

README.md

Alternative: kafka-to-volume (no ngrok required)

Databricks Serverless cannot reach a local ngrok Kafka endpoint. Use the built-in kafka-to-volume Docker profile to upload CDC events to a Databricks Volume landing zone, allowing Bronze to use Auto Loader instead of direct Kafka connectivity.

docker compose --profile kafka-to-volume up -d kafka-to-volume

Source Tables (PostgreSQL dvdrental — all 15)

Note: When using local ngrok for Kafka connectivity, 3 tables (category, country, film_actor) may not reach Bronze due to Databricks Serverless network limitations. Use the kafka-to-volume Docker profile as a workaround (see Architecture section).

Reference / Dimension

Table	Notes
public.country	Lookup
public.city	→ country
public.address	→ city
public.language	Film language lookup
public.category	Film category lookup
public.actor	Actor dimension
public.store	Store dimension
public.staff	Staff dimension → address, store
public.customer	Customer dimension → address, store

Transaction / Fact

Table	Changes captured
film	rental_rate, rental_duration, replacement_cost
film_actor	Junction: film ↔ actor
film_category	Junction: film ↔ category
inventory	Film copies per store
rental	New rentals, return_date updates
payment	Payment inserts

---

Databricks Jobs

Job	ID	Tasks	Description
dvdrental-bronze	325293262130713	1	Kafka → Bronze Delta (availableNow trigger)
dvdrental-silver	1099814608698427	15	Bronze → Silver MERGE, 3 batches of 5
dvdrental-vault	950203691556666	4	Hubs → Links+Sats → Business Vault
dvdrental-vault-gold	83436339832760	1	dbt build vault+gold via NB_run_dbt
dvdrental-orchestrator	684287727358557	4	Chains: bronze → silver → vault → vault-gold
pumpapi-lakehouse	deployed via databricks bundle deploy	1 pipeline	Lakeflow: Bronze + 3 Silver tables
pumpfun-bronze	deployed via databricks bundle deploy	1	Triggers pumpapi-lakehouse, every 2 min

---

pump.fun Pipeline

Independent of the dvdrental CDC pipeline above. The producer (ingestion/pumpfun/) is a standalone websocket service — it does not run as a Databricks job, since it needs to be always-on rather than scheduled. Each batch is compressed (zstd) and base64-encoded before upload:

cd ingestion/pumpfun
cp .env.example .env   # fill in DATABRICKS_HOST / DATABRICKS_TOKEN
docker compose up -d --build   # or: python3 -m venv venv && pip install -r requirements.txt && python -m app.main

See ingestion/pumpfun/README.md for running it as a systemd service instead. Once it's landing files in the Volume, run:

cd orchestration/bundle && databricks bundle deploy -t dev

to deploy pumpapi-lakehouse — a single Lakeflow Declarative Pipeline (pyspark.pipelines, see pumpapi-lakehouse/README.md) that does Bronze + all 3 Silver tables — and the pumpfun-bronze job that triggers it every 2 minutes. The Pipelines API doesn't support the git_source mechanism scripts/deploy_jobs.py uses for plain notebook jobs, so that script manages no pump.fun resources at all.

---

Local Infrastructure

# Start the full CDC stack (Zookeeper, Kafka, PostgreSQL 15, Debezium Connect, Schema Registry, Kafka UI)
docker compose up -d

# Register the Debezium connector (wait ~30s for Kafka Connect to be ready)
curl -X POST http://localhost:8083/connectors \
  -H 'Content-Type: application/json' \
  --data @postgres-connector.json

# Check connector status
curl http://localhost:8083/connectors/postgres-connector/status

# Kafka UI
open http://localhost:8085

# Alternative: kafka-to-volume (no ngrok needed for Serverless workspaces)
docker compose --profile kafka-to-volume up -d kafka-to-volume

---

Python Setup

pip install -r requirements.txt

# Copy and fill in Databricks credentials
cp .envexample .env
# Edit .env: set DATABRICKS_HOST and DATABRICKS_TOKEN

---

Data Generators

# Bulk seed: inserts 1000 customers, 1000 films, 10000+ DML events (run once before pipeline)
python3 ingestion/generators/load_bulk_data.py

# Film generator (updates rental_rate, rental_duration, replacement_cost on existing films)
python3 ingestion/generators/load_products_generator.py

# Rental + payment generator (new rentals, film returns, payments)
python3 ingestion/generators/load_generator.py

# Optional env vars: ITERATIONS, SLEEP_MIN, SLEEP_MAX
# DB env vars: PGHOST, PGPORT, PGDATABASE, PGUSER, PGPASSWORD

---

Deployment

set -a && source .env && set +a

# Push Kafka credentials to Databricks secret scope
python3 scripts/push_secrets_to_databricks.py

# Upload vault config to Unity Catalog Volume (required before first vault run)
python3 scripts/upload_vault_config.py

# Deploy all Databricks jobs (dvdrental + pumpfun)
python3 scripts/deploy_jobs.py

# Deploy and immediately trigger the full orchestrator run
python3 scripts/deploy_jobs.py --run-orchestrator

---

dbt (Vault + Gold Layer)

# Local
cd transformation/dbt_project
dbt debug          # verify connection
dbt build --select vault gold   # run vault + gold models + tests
dbt test           # data quality tests only

# On Databricks: run transformation/NB_run_dbt.ipynb (uses dbtRunner API, no subprocess)

---

Run the Pipeline

1. Start local infrastructure: docker compose up -d
2. Register the Debezium connector
3. Seed bulk data: python3 ingestion/generators/load_bulk_data.py
4. Deploy jobs: python3 scripts/deploy_jobs.py
5. Trigger orchestrator from the Databricks UI or: python3 scripts/deploy_jobs.py --run-orchestrator

The orchestrator runs: Bronze → Silver → Vault → Vault-Gold (dbt)