dlt-pipeline

DLT Pipeline Patterns — Alpura Medallion Architecture

Apply to every DLT pipeline notebook. Pipelines always flow Bronze → Silver → Gold. Never skip Silver. Never write business logic in Bronze.

Rules

• Bronze: raw ingestion via Auto Loader — no transformations, no filtering
• Silver: validated, deduped, cleaned — @dlt.expect_or_drop on every quality dimension
• Gold: aggregated, app-ready, business-metric-named — @dlt.materialized_view preferred
• Never skip Silver — no Bronze → Gold directly
• Always use Databricks Volumes (/Volumes/catalog/schema/path) — never dbfs:/
• Use dlt.read() / dlt.read_stream() — never spark.read() inside DLT
• Name tables: bronze_<source>, silver_<entity>, gold_<metric>
• Log ingestion count and key metrics at every layer — use _logger.py
• All pipeline code runs in DLT notebooks — never in regular notebooks
• Schema evolution: use cloudFiles.schemaEvolutionMode = "rescue" at Bronze

Bronze — Auto Loader Ingestion

JSON / CSV from Volumes

import dlt
from pyspark.sql import functions as F
from _logger import get_logger
logger = get_logger(__name__)

@dlt.table(
    name="bronze_sales_orders",
    comment="Raw sales orders — Auto Loader ingestion from Volumes landing zone",
    table_properties={"quality": "bronze", "pipelines.reset.allowed": "true"},
)
def bronze_sales_orders():
    logger.info("Ingesting bronze_sales_orders")
    return (
        spark.readStream.format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.schemaLocation",  "/Volumes/prod/raw/checkpoints/sales_orders_schema")
        .option("cloudFiles.schemaEvolutionMode", "rescue")   # new columns → _rescued_data
        .option("cloudFiles.inferColumnTypes", "true")
        .load("/Volumes/prod/raw/landing/sales_orders/")
        .withColumn("_ingested_at",  F.current_timestamp())
        .withColumn("_source_file",  F.input_file_name())
    )

Parquet / Delta from Volumes