Abhishek.
All projects

2025

Event-Driven Lakehouse

A local-first, backend-agnostic medallion lakehouse with queue-driven ingestion. Reliably ingests irregularly-arriving files with exactly-once processing and schema enforcement across raw → aggregate → curated layers. A queue layer (local watcher or Azure Storage Queue) feeds N async workers with in-flight deduplication, while a single-writer commit buffer prevents Delta Lake write contention and enables near-linear worker scaling. YAML-defined schemas mean zero-Python table definitions, and it runs without a JVM or Spark — querying via DuckDB locally or scaling to ADLS Gen2 in the cloud.

PythonDelta Lake (delta-rs)DuckDBasyncioAzure Storage QueueADLS Gen2
View on GitHub

event-driven-lakehouse

a local-first, backend-agnostic medallion lakehouse with queue-driven ingestion.

The problem

Files arrive in a landing zone at irregular intervals — a batch every few minutes, a burst of hundreds at once, or a trickle of one per hour. You need:

  • Exactly-once ingestion — no duplicated rows if a message is redelivered.
  • Schema enforcement — every column cast to the declared type before writing.
  • A layered lake — raw (file-grained), aggregate (date-partitioned), and curated (upsert by business key) so consumers always read a clean, consistent view.
  • Zero ops overhead — it must run on a laptop for development and testing, then scale to cloud storage with only an environment-variable change.

Architecture

  ┌──────────────┐
  │  File Drop   │  (CSV / Parquet landing zone)
  │  Directory   │
  └──────┬───────┘
         │ file-arrival event
         ▼
  ┌──────────────┐
  │    Queue     │  local: FileDropQueue  │  cloud: Azure Storage Queue
  └──────┬───────┘
         │ messages (batch receive, visibility timeout)
         ▼
  ┌──────────────────────────────────┐
  │   N Async Workers (parse/cast)   │
  │   in-flight dedup by file_key    │
  └──────────────┬───────────────────┘
                 │ Arrow Tables
                 ▼
  ┌──────────────────────────────────┐
  │   Single-Writer Commit Buffer    │  (flush on rows or interval)
  └──────────────┬───────────────────┘
                 │
                 ▼
  ┌──────────────────────────────────┐
  │        raw Delta Table           │  append-only, file-grained
  └──────────────┬───────────────────┘
                 │
                 ▼
  ┌──────────────────────────────────┐
  │     aggregate Delta Table        │  append, partitioned by date
  └──────────────┬───────────────────┘
                 │
                 ▼
  ┌──────────────────────────────────┐
  │      curated Delta Table         │  upsert by primary keys
  └──────────────┬───────────────────┘
                 │
        ┌────────┴────────┐
        ▼                 ▼
   DuckDB SQL           BI Tools
   (local)             (Power BI, etc.)

Design decisions & trade-offs

Buffered single-writer commit

Delta Lake uses optimistic concurrency: every writer reads the current log version, writes parquet, then tries to commit a new log entry. When N workers all try to commit simultaneously they race, and all but one must retry — at high throughput this burns CPU and causes cascading delays.

The solution here is a single writer coroutine. All N workers parse and cast data concurrently (the CPU-bound part), then push Arrow tables onto an internal asyncio queue. One writer drains that queue and commits one Delta transaction per flush. This gives near-linear worker scaling without commit contention. The trade-off is that the writer is the throughput bottleneck for very large batches; in practice the per-commit overhead is small compared to parse/cast time.

YAML schemas for the raw layer

Rather than defining column types in Python per table, each table has a YAML file declaring column names, types, nullable flags, max lengths, primary keys, and write mode. This has three benefits:

  1. Adding a new table requires zero Python — drop a YAML file.
  2. The schema loader is LRU-cached, so it is read once per process.
  3. YAML diffs are easy to review in pull requests.

The trade-off is that YAML is less expressive than Python for complex conditional logic. For the raw layer this is intentional — raw ingestion should be dumb. Business logic lives in the transform registry, not the schema.

Local-first with pluggable cloud backends

The two backends — STORAGE_BACKEND (local Delta vs. ADLS Gen2) and QUEUE_BACKEND (file-drop watcher vs. Azure Storage Queue) — are selected at runtime via environment variables. All Azure imports are lazy: they only execute when the azure backend is chosen, so missing Azure extras never break a local run.

This makes the local development loop instant: python demo.py with no credentials produces a fully-functional lakehouse in ./data/. Switching to cloud is one export STORAGE_BACKEND=azure away.

Why delta-rs (not Spark)

delta-rs is a Rust implementation of the Delta Lake protocol with Python bindings. It writes valid Delta tables with no JVM, no Spark cluster, and no Java dependency. The same parquet files and _delta_log entries are readable by Spark, Databricks, and any other Delta-compatible reader. For file-at-a-time ingestion on a laptop or a small VM, the per-commit latency of delta-rs is typically 10–50 ms — far below the cost of a Spark job startup.

Visibility-timeout retries vs. an explicit DLQ

When a worker fails to process a message, it calls nack() which makes the message visible again immediately. The queue re-delivers it after the visibility timeout expires if nack() is not called (e.g., on a crash). This means every message gets at least one retry automatically with no extra infrastructure.

The trade-off is that a poison-pill file (one that always fails to parse) will loop forever. In production, combine this with a dequeue-count threshold: after N redeliveries write the message to an error log and stop retrying. The processed-files ledger in maintenance/checkpoint.py enables this pattern.

In-flight dedup vs. the processed-files ledger

The in-flight dedup set (table, file_key) prevents a message from being processed twice while it is already in the pipeline within the same process run. It is best-effort: if the process restarts mid-batch the dedup set is lost.

For hard exactly-once semantics, combine in-flight dedup with the processed-files ledger: before writing to raw, check whether the file_key appears in the ledger with status=ok. If it does, skip it. The ledger is itself a Delta table, so this check is transactionally consistent.

Quickstart

git clone https://github.com/shrishri108/event-driven-lakehouse
cd event-driven-lakehouse
uv sync
python demo.py

Expected output (abbreviated):

============================================================
  1. Initialising data directories
============================================================
  --> Clean slate at .../data

============================================================
  2. Generating synthetic trip data
============================================================
  --> Generated 5 files x 1000 rows in .../data/queue/incoming

============================================================
  4. Running poller (4 workers) -> raw Delta
============================================================
  --> Ingested 5000 rows into raw/trips in 4.2s

============================================================
  5. Running medallion pipeline (raw -> aggregate -> curated)
============================================================
  --> aggregate/trips:       35 rows
  --> curated/trips_daily:   35 rows

============================================================
  6. Querying via DuckDB
============================================================
  --> DuckDB raw/trips row count:  5000
  --> Top 5 curated rows (by revenue):
      (datetime.date(2024, 1, 7), 'Midtown', 42, Decimal('4821.30'))
      ...

============================================================
  Done in 18.3s
============================================================

Optional Azure "production mode"

Install Azure extras:

uv sync --extra azure

Configure environment variables (copy .env.example to .env and fill in):

export STORAGE_BACKEND=azure
export QUEUE_BACKEND=azure
export AZURE_STORAGE_ACCOUNT=myaccount
export AZURE_STORAGE_CONTAINER=mylake
export AZURE_QUEUE_NAME=myqueue
export AZURE_QUEUE_ACCOUNT_URL=https://myaccount.queue.core.windows.net

Authenticate via any DefaultAzureCredential flow:

  • Local dev: az login
  • CI / VM: set AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, AZURE_TENANT_ID
  • Azure compute: managed identity (no env vars needed)

Then run python demo.py — the same code path uses ADLS Gen2 instead of the local filesystem.

Project layout

event-driven-lakehouse/
  demo.py                  one-command end-to-end demo
  pyproject.toml           uv-managed, Python 3.12
  schemas/
    raw/trips.yaml         taxi-style raw schema
    curated/trips_daily.yaml  daily aggregate schema
  sample_data/
    generate.py            synthetic data generator
  src/lakehouse/
    config.py              env-driven backend factories
    ingest/                queue adapters + async poller
    storage/               Delta store adapters
    schema/                YAML loader + Arrow type caster
    transform/             medallion pipeline + transform registry
    maintenance/           dedup, compaction, z-order, orphan GC, repair
    sql/                   DuckDB engine
    concurrency/           multiprocess log relay + process pool
    utils/                 bulk loader
    bench/                 throughput measurement
  tests/                   pytest suite

Tests

uv run pytest
# or
python -m pytest

Maintenance

Delta tables accumulate small files, stale parquet, and duplicate rows over time. The lakehouse.maintenance package provides table-level operations to keep the lake healthy. Every function takes a DeltaStore (obtained from config.get_storage()) and a table_name like "raw/trips" or "curated/trips_daily", and is safe to call against either the local or Azure backend.

from lakehouse.config import get_storage

store = get_storage()

Compaction & vacuum

Coalesce small parquet files into larger ones and remove files no longer referenced by the Delta log. Run this regularly on append-heavy tables (raw and aggregate) where each ingested file produces a separate part file.

from lakehouse.maintenance.compaction import compact, vacuum, compact_and_vacuum

# Compact small files into ~128 MiB targets.
compact(store, "raw/trips", target_size=128 * 1024 * 1024)

# Delete unreferenced parquet older than the retention window (default 7 days).
# Use dry_run=True first to preview what would be deleted.
vacuum(store, "raw/trips", retention_hours=168, dry_run=True)

# Convenience: compact then vacuum in one call.
compact_and_vacuum(store, "raw/trips")

Caution: vacuum permanently removes old file versions, which disables time travel to versions whose files have been purged. Keep retention_hours at least as long as your longest-running reader.

Z-ordering

Cluster data on the columns you filter by most often to improve data-skipping. Run after large appends or whenever query patterns shift.

from lakehouse.maintenance.zorder import zorder

zorder(store, "curated/trips_daily", ["pickup_zone"])

Deduplication

Remove exact duplicate rows by SHA-256 row hash, keeping the first occurrence. This rewrites the table (overwrite) and verifies the resulting row count.

from lakehouse.maintenance.dedup import dedup_table

stats = dedup_table(store, "raw/trips")
# -> {"before": 5000, "after": 4998, "removed": 2}

Orphan & missing-file repair

Reconcile the filesystem with the Delta transaction log. remove_orphans deletes parquet files on disk that the log no longer references; repair_missing_files rewrites the table to drop log references whose parquet files have gone missing (e.g. after a partial restore). Both accept dry_run=True.

from lakehouse.maintenance.orphan_gc import remove_orphans
from lakehouse.maintenance.repair import repair_missing_files

remove_orphans(store, "raw/trips", dry_run=True)       # list orphaned files
repair_missing_files(store, "raw/trips", dry_run=True)  # list missing references

Checkpointing & the processed-files ledger

delta-rs auto-checkpoints every 10 commits; force one on demand to speed up log replay, and inspect the ingestion ledger for observability.

from lakehouse.maintenance.checkpoint import force_checkpoint, read_ledger

force_checkpoint(store, "raw/trips")

ledger = read_ledger(store)   # Arrow table of every ingested file_key + status

Time travel

Read a table as it looked at a past version or point in time, or list its commit history.

from datetime import datetime, timezone
from lakehouse.maintenance.timetravel import load_version, load_at, list_versions

list_versions(store, "raw/trips")                       # commit history
load_version(store, "raw/trips", version=3)             # snapshot at v3
load_at(store, "raw/trips", datetime(2024, 1, 7, tzinfo=timezone.utc))

Suggested cadence

OperationWhen to run
compact / vacuumDaily on append-heavy raw & aggregate tables
zorderAfter large appends or query-pattern changes
dedup_tableAd hoc, if duplicates are suspected
remove_orphansAfter interrupted writes or restores
repair_missing_filesAfter a partial restore / missing parquet
force_checkpointBefore archival or to speed up log replay

demo.py step 7 runs compact_and_vacuum + zorder on curated/trips_daily as a worked example.

License

MIT — see LICENSE.