← All posts

Databricks Delta Live Tables: Declarative Pipeline Orchestration Made Simple

6 min read5.1k
#Databricks#Python#ETL#DataEngineering#DeltaLake#Spark

If you've ever spent more time debugging broken Spark jobs than actually building pipelines, Databricks Delta Live Tables (DLT) is worth your attention. Instead of writing imperative Spark code that moves data from A to B and hoping nothing breaks in between, DLT lets you declare what your tables should look like and Databricks figures out the rest — dependencies, retries, data quality checks, and all.

In this post I'll walk through building a real-world DLT pipeline from scratch on Databricks, covering everything from raw ingestion to serving-ready Gold tables with data quality expectations baked in.

What is Databricks Delta Live Tables?

DLT is a declarative framework built on top of Delta Lake and Apache Spark. Instead of writing spark.read → transform → spark.write, you define tables using @dlt.table decorators and DLT handles:

  • Dependency resolution and execution order
  • Incremental vs full refresh modes
  • Data quality enforcement via Expectations
  • Automatic lineage tracking
  • Pipeline monitoring and alerting

The result is pipelines that are easier to read, easier to test, and dramatically easier to recover when something goes wrong.

Architecture Overview

Pipeline Flow

DLT vs Traditional Spark Pipelines

FeatureTraditional SparkDatabricks Delta Live Tables
Dependency managementManual, error proneAutomatic via DAG resolution
Data qualityAd-hoc checks in codeDeclarative Expectations
Incremental processingManual watermark logicBuilt-in STREAMING tables
Retries and recoveryCustom retry wrappersManaged by DLT runtime
Lineage trackingNone out of the boxAutomatic in Unity Catalog
Pipeline monitoringCloudWatch / customBuilt-in DLT event log
Schema evolutionManual mergeSchemaHandled automatically

Step 1 — Bronze: Raw Ingest with Auto Loader

Auto Loader is DLT's preferred ingest mechanism. It tracks which files have been processed using checkpoints so you never double-ingest.

python
import dlt
from pyspark.sql.functions import current_timestamp, input_file_name

@dlt.table(
    name="bronze_orders",
    comment="Raw order events ingested from ADLS Gen2",
    table_properties={"quality": "bronze"}
)
def bronze_orders():
    return (
        spark.readStream
            .format("cloudFiles")
            .option("cloudFiles.format", "json")
            .option("cloudFiles.schemaLocation", "/checkpoints/orders/schema")
            .load("abfss://raw@yourstorage.dfs.core.windows.net/orders/")
            .withColumn("_ingested_at", current_timestamp())
            .withColumn("_source_file", input_file_name())
    )

Step 2 — Silver: Clean Data with Expectations

Expectations are DLT's data quality layer. You define rules and tell DLT what to do when rows violate them: warn, drop, or fail the pipeline.

python
import dlt
from pyspark.sql.functions import col, to_timestamp, upper, trim

# Define reusable expectations
ORDER_EXPECTATIONS = {
    "valid_order_id":    "order_id IS NOT NULL",
    "valid_customer_id": "customer_id IS NOT NULL",
    "valid_amount":      "order_amount > 0",
    "valid_status":      "status IN ('PENDING', 'CONFIRMED', 'SHIPPED', 'CANCELLED')",
}

@dlt.table(
    name="silver_orders",
    comment="Cleaned and validated order records",
    table_properties={"quality": "silver"}
)
@dlt.expect_all_or_drop(ORDER_EXPECTATIONS)   # drop rows that fail any rule
def silver_orders():
    return (
        dlt.read_stream("bronze_orders")
            .withColumn("order_ts",   to_timestamp(col("order_timestamp")))
            .withColumn("status",     upper(trim(col("status"))))
            .withColumn("country",    upper(trim(col("country_code"))))
            .filter(col("order_id").isNotNull())
            .select(
                "order_id", "customer_id", "order_ts",
                "order_amount", "status", "country",
                "product_id", "_ingested_at"
            )
    )

# Quarantine table — keep dropped rows for audit
@dlt.table(
    name="silver_orders_quarantine",
    comment="Rows that failed data quality expectations"
)
@dlt.expect_all_or_drop({v: f"NOT ({v})" for v in ORDER_EXPECTATIONS.values()})
def silver_orders_quarantine():
    return dlt.read_stream("bronze_orders")

Step 3 — Gold: Aggregated Business Table

Gold tables are materialized views over Silver. In DLT you define them the same way — just read from Silver instead of Bronze.

python
import dlt
from pyspark.sql.functions import col, count, sum as _sum, avg, countDistinct, date_trunc

@dlt.table(
    name="gold_daily_order_summary",
    comment="Daily aggregated order metrics per country and status",
    table_properties={"quality": "gold"}
)
def gold_daily_order_summary():
    return (
        dlt.read("silver_orders")
            .withColumn("order_date", date_trunc("day", col("order_ts")))
            .groupBy("order_date", "country", "status")
            .agg(
                count("order_id")           .alias("total_orders"),
                _sum("order_amount")        .alias("total_revenue"),
                avg("order_amount")         .alias("avg_order_value"),
                countDistinct("customer_id").alias("unique_customers"),
            )
            .orderBy("order_date", "country")
    )

@dlt.table(
    name="gold_customer_lifetime_value",
    comment="Cumulative LTV per customer across all orders"
)
def gold_customer_lifetime_value():
    return (
        dlt.read("silver_orders")
            .filter(col("status") != "CANCELLED")
            .groupBy("customer_id")
            .agg(
                count("order_id")    .alias("total_orders"),
                _sum("order_amount") .alias("lifetime_value"),
                avg("order_amount")  .alias("avg_order_value"),
            )
    )

Step 4 — Pipeline Configuration

DLT pipelines are configured via JSON or the Databricks UI. Here's the JSON config you'd use with the CLI or Terraform.

json
{
  "name": "orders-dlt-pipeline",
  "target": "orders_catalog.dlt",
  "libraries": [
    { "notebook": { "path": "/pipelines/orders_pipeline" } }
  ],
  "clusters": [
    {
      "label": "default",
      "autoscale": {
        "min_workers": 2,
        "max_workers": 8,
        "mode": "ENHANCED"
      }
    }
  ],
  "continuous": false,
  "development": false,
  "edition": "ADVANCED",
  "photon": true,
  "configuration": {
    "pipelines.enableTrackHistory": "true"
  }
}

Things to Watch in Production

Use ADVANCED edition for Expectations. The CORE edition doesn't support data quality Expectations. Make sure your pipeline is set to ADVANCED or you'll get a silent no-op on your quality rules.

Quarantine bad rows, don't just drop them. @dlt.expect_all_or_drop silently removes failing rows. Always pair it with a quarantine table so you have an audit trail of what was dropped and why.

Streaming tables vs materialized views. Use @dlt.table with spark.readStream for Bronze and Silver (incremental). Use @dlt.table with dlt.read (batch) for Gold aggregations. Mixing them incorrectly causes full re-scans on every pipeline run.

Event log is queryable. DLT writes all pipeline events to a Delta table at <storage_location>/system/events. Query it directly to build custom monitoring dashboards in Databricks SQL.

Wrapping Up

Databricks Delta Live Tables removes the scaffolding that makes Spark pipelines painful: manual dependency management, ad-hoc quality checks, and custom retry logic. What you're left with is clean declarative code where every table has a clear definition, a quality contract, and an automatic audit trail.

The Medallion Architecture maps perfectly onto DLT — Bronze for raw ingest, Silver for quality enforcement, Gold for business aggregations — and the combination of Auto Loader and Expectations makes it production-ready with surprisingly little boilerplate.

References