Building Production-Grade Delta Lake Pipelines With Apache Spark on Databricks

Why Delta Lake?

Apache Parquet on cloud storage was a great first step for data lakes — but it left engineers dealing with a painful set of problems in production:

• No ACID transactions — concurrent reads/writes could corrupt data silently
• Schema drift — nothing stopped upstream systems from changing column types
• No deletes or updates — GDPR compliance meant rewriting entire partitions
• Painful failure recovery — half-written data after a job crash became your problem

Delta Lake solves all of this by sitting on top of Parquet and adding a transaction log (_delta_log/) that records every operation atomically. On Databricks, Delta is the default table format, deeply integrated with Apache Spark, Auto Optimize, and the Photon execution engine.

The Medallion Architecture

The medallion (or multi-hop) architecture organizes data into three progressive refinement layers. Each layer has a clear contract with the layers around it.

Each layer has a distinct responsibility:

The key design principle is: never skip a layer. Debugging a production incident is infinitely easier when you can replay from raw Bronze data.

Delta Lake Internals: The Transaction Log

Before writing a single line of code, it's worth understanding how Delta Lake achieves ACID semantics. Every write operation (INSERT, UPDATE, DELETE, MERGE) produces a new JSON entry in _delta_log/.

This log-based approach gives you time travel for free — querying VERSION AS OF 2 simply replays only the log entries up to that point. Checkpoints every 10 commits keep read performance snappy even with thousands of versions.

Setting Up Your Databricks Environment

All code here runs on Databricks Runtime 13.x+ (which ships Delta Lake 2.x). For local dev, use the delta-spark package.

# Databricks notebook — Runtime 13.3 LTS or higher
# Delta Lake is pre-installed; no pip install needed

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import (
    StructType, StructField,
    StringType, LongType, DoubleType, TimestampType, BooleanType
)
from delta.tables import DeltaTable

# On Databricks, SparkSession is pre-created as `spark`
# For local testing:
# spark = (SparkSession.builder
#     .appName("medallion-pipeline")
#     .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
#     .config("spark.sql.catalog.spark_catalog", "spark.sql.delta.catalog.DeltaCatalog")
#     .getOrCreate())

# Unity Catalog paths (recommended for production)
CATALOG   = "prod"
BRONZE_DB = f"{CATALOG}.bronze"
SILVER_DB = f"{CATALOG}.silver"
GOLD_DB   = f"{CATALOG}.gold"

# DBFS / external storage path (for raw file landing)
RAW_LANDING = "abfss://[email protected]/events/"

Building the Bronze Layer

Bronze is your append-only raw ingestion layer. The goal is landing data with zero transformation — preserve everything, even malformed records. Schema is inferred or declared loosely.

# ── Bronze Ingestion ──────────────────────────────────────────────────────────
# Using Auto Loader (cloudFiles) — the Databricks-native incremental ingest
# mechanism. It tracks which files have been processed via a checkpoint,
# so re-running never double-injects data.

raw_event_schema = StructType([
    StructField("event_id",        StringType(),    True),
    StructField("user_id",         StringType(),    True),
    StructField("event_type",      StringType(),    True),
    StructField("event_ts",        StringType(),    True),  # keep as string in bronze
    StructField("properties",      StringType(),    True),  # raw JSON blob
    StructField("session_id",      StringType(),    True),
    StructField("platform",        StringType(),    True),
])

bronze_stream = (
    spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.schemaLocation", "/checkpoints/bronze_events_schema")
        .option("cloudFiles.inferColumnTypes", "false")   # keep raw types
        .schema(raw_event_schema)
        .load(RAW_LANDING)
        # Enrich with ingestion metadata — critical for debugging
        .withColumn("_ingest_ts",       F.current_timestamp())
        .withColumn("_source_file",     F.input_file_name())
        .withColumn("_ingest_date",     F.to_date(F.current_timestamp()))
)

(
    bronze_stream.writeStream
        .format("delta")
        .outputMode("append")
        .option("checkpointLocation", "/checkpoints/bronze_events")
        .option("mergeSchema", "true")          # allow new columns from upstream
        .partitionBy("_ingest_date")            # partition for incremental silver reads
        .trigger(availableNow=True)             # run-once trigger (for scheduled jobs)
        .tableCheckpoint(f"{BRONZE_DB}.events_raw")
        .toTable(f"{BRONZE_DB}.events_raw")
)

Pro tip: Always add _ingest_ts and _source_file metadata columns in Bronze. When an upstream system sends corrupt data at 3 AM, these columns tell you exactly which file batch caused it.

Transforming to Silver

Silver is where the real engineering work happens: deduplication, type casting, schema validation, and applying business rules. We also handle CDC (Change Data Capture) upserts here using Delta's MERGE.

# ── Silver Transformation ─────────────────────────────────────────────────────

def transform_bronze_to_silver(bronze_df):
    """
    Apply cleansing and conforming rules to raw Bronze events.
    Returns a Silver-ready DataFrame with enforced schema.
    """
    return (
        bronze_df
            # 1. Parse timestamps properly
            .withColumn("event_ts", F.to_timestamp("event_ts", "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"))

            # 2. Parse the raw JSON properties blob into a struct
            .withColumn("props", F.from_json(
                F.col("properties"),
                schema="page_url STRING, referrer STRING, duration_ms LONG, revenue DOUBLE"
            ))
            .drop("properties")

            # 3. Normalize platform values
            .withColumn("platform", F.lower(F.trim(F.col("platform"))))
            .withColumn("platform", F.when(
                F.col("platform").isin("ios", "android"), F.col("platform")
            ).when(
                F.col("platform").isin("web", "browser", "desktop"), F.lit("web")
            ).otherwise(F.lit("unknown")))

            # 4. Filter out test/internal traffic
            .filter(~F.col("user_id").startswith("test_"))
            .filter(F.col("event_id").isNotNull())

            # 5. Deduplicate within the micro-batch (window by event_id)
            .dropDuplicates(["event_id"])

            # 6. Add Silver metadata
            .withColumn("_silver_ts",   F.current_timestamp())
            .withColumn("event_date",   F.to_date("event_ts"))   # partition key
    )


# Upsert into Silver using MERGE (handles late-arriving / duplicate events)
def upsert_to_silver(micro_batch_df, batch_id):
    micro_batch_df = transform_bronze_to_silver(micro_batch_df)

    silver_table = DeltaTable.forName(spark, f"{SILVER_DB}.events_clean")

    (
        silver_table.alias("target")
            .merge(
                micro_batch_df.alias("source"),
                "target.event_id = source.event_id"   # dedup key
            )
            .whenMatchedUpdateAll()       # update if record arrived late with corrections
            .whenNotMatchedInsertAll()    # insert new records
            .execute()
    )


# Stream from Bronze → Silver using foreachBatch
(
    spark.readStream
        .format("delta")
        .option("readChangeFeed", "true")    # CDF — only process new Bronze rows
        .table(f"{BRONZE_DB}.events_raw")
        .writeStream
        .foreachBatch(upsert_to_silver)
        .option("checkpointLocation", "/checkpoints/silver_events")
        .trigger(availableNow=True)
        .start()
)

Aggregating to Gold

Gold tables are business-ready aggregates consumed directly by BI tools, dashboards, and ML feature pipelines. They are typically batch-refreshed on a schedule.

# ── Gold Aggregation ──────────────────────────────────────────────────────────

daily_revenue = (
    spark.table(f"{SILVER_DB}.events_clean")
        .filter(F.col("event_type") == "purchase")
        .filter(F.col("event_date") >= F.date_sub(F.current_date(), 90))  # rolling 90d
        .groupBy("event_date", "platform")
        .agg(
            F.sum("props.revenue").alias("total_revenue"),
            F.countDistinct("user_id").alias("unique_buyers"),
            F.count("event_id").alias("transaction_count"),
            F.avg("props.duration_ms").alias("avg_session_duration_ms"),
        )
        .withColumn("revenue_per_buyer",
            F.round(F.col("total_revenue") / F.col("unique_buyers"), 2))
        .withColumn("_gold_ts", F.current_timestamp())
)

# Overwrite with replaceWhere — only touch the last 90 days, not the full table
(
    daily_revenue.write
        .format("delta")
        .mode("overwrite")
        .option("replaceWhere", "event_date >= date_sub(current_date(), 90)")
        .saveAsTable(f"{GOLD_DB}.daily_revenue")
)

Z-Ordering and Data Skipping

Z-ordering is Databricks' multi-dimensional clustering technique. It co-locates related data within the same set of Parquet files, so Spark can skip irrelevant files entirely during queries — without the overhead of strict partitioning.

-- Run OPTIMIZE + ZORDER after significant writes
-- This rewrites data files to cluster on the most commonly filtered columns

OPTIMIZE prod.silver.events_clean
ZORDER BY (user_id, event_date, event_type);

-- Check how many files were skipped in your last query
-- (run immediately after a SELECT with filters)
SELECT
    operation,
    operationMetrics['numFilesAdded']    AS files_added,
    operationMetrics['numFilesRemoved']  AS files_removed,
    operationMetrics['numRemovedBytes']  AS bytes_removed
FROM (
    DESCRIBE HISTORY prod.silver.events_clean
)
WHERE operation = 'OPTIMIZE'
ORDER BY timestamp DESC
LIMIT 5;

Rule of thumb: Z-order on your top 3–4 most-filtered columns. Beyond that, the clustering benefit diminishes and OPTIMIZE runtimes grow significantly. Never Z-order on partition columns — they're already physically separated.

Optimized Spark Writes

Poorly tuned Spark writes are the #1 cause of small-file problems in Delta Lake. Here's a production-hardened write configuration:

# ── Write Configuration Reference ────────────────────────────────────────────

SILVER_WRITE_CONFIG = {
    # Coalesce output files to ~128MB each (avoids small-file explosion)
    "spark.sql.shuffle.partitions": "200",          # tune to cluster size
    "spark.databricks.delta.optimizeWrite.enabled": "true",   # auto bin-packing
    "spark.databricks.delta.autoCompact.enabled": "true",      # background compaction

    # Target file size for Auto Optimize
    "spark.databricks.delta.optimizeWrite.binSize": "134217728",  # 128 MB in bytes

    # Enable deletion vectors (Databricks 12.2+) — soft-deletes without file rewrites
    "spark.databricks.delta.enableDeletionVectors": "true",
}

# Apply at the session level for the pipeline job
for k, v in SILVER_WRITE_CONFIG.items():
    spark.conf.set(k, v)


# For partitioned tables: control output file count per partition
(
    silver_df
        .repartition(F.col("event_date"))          # one task group per date partition
        .write
        .format("delta")
        .mode("overwrite")
        .option("dataChange", "true")
        .option("overwriteSchema", "false")        # never silently change schema in prod
        .partitionBy("event_date")
        .saveAsTable(f"{SILVER_DB}.events_clean")
)

Pipeline Comparison Table

Here's how different ingestion patterns stack up for common production scenarios on Databricks:

DLT note: Delta Live Tables is Databricks' managed pipeline framework that handles the orchestration, monitoring, and retry logic described above declaratively. For teams starting fresh, DLT is worth evaluating before building the plumbing manually.

Key Takeaways

• Medallion architecture separates concerns cleanly: Bronze for fidelity, Silver for correctness, Gold for consumption.
• Delta's transaction log is the foundation of all ACID guarantees — understanding it helps you debug merge conflicts, time travel, and VACUUM safely.
• Auto Loader is the right default for cloud file ingestion on Databricks — it handles exactly-once semantics and schema evolution automatically.
• MERGE with foreachBatch is the idiomatic pattern for deduplication and CDC in Spark Structured Streaming.
• Z-ORDER + Auto Optimize should be standard practice for Silver and Gold tables that receive frequent queries with selective filters.
• Deletion Vectors (Databricks 12.2+) make point deletes significantly cheaper — enable them for tables with GDPR or compliance requirements.

References

1. Delta Lake Documentation — Delta Lake Transaction Log — The official deep dive into how the _delta_log works internally.
2. Databricks — Medallion Architecture
3. Databricks — Auto Loader (cloudFiles)
4. Databricks — Delta Lake OPTIMIZE and Z-Ordering
5. Databricks — Auto Optimize (Optimized Writes + Auto Compaction)
6. Databricks — Deletion Vectors
   
7. Databricks — Delta Live Tables Overview
8. Structured Streaming + foreachBatch — Apache Spark Docs
9. "The Delta Lake Paper" — VLDB 2020 (Armbrust et al.)
10. Databricks Blog — Diving Into Delta Lake: Unpacking the Transaction Log