Delta Live Tables & Declarative Pipelines

# Traditional imperative pipeline
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, when, current_timestamp

spark = SparkSession.builder.getOrCreate()

# Step 1: Read raw data
raw_df = spark.read.format("json").load("/data/raw/events/")

# Step 2: Clean and validate
cleaned_df = raw_df.filter(col("event_id").isNotNull())
cleaned_df = cleaned_df.dropDuplicates(["event_id"])

# Step 3: Transform
enriched_df = cleaned_df.withColumn(
    "processed_at", current_timestamp()
)

# Step 4: Write (handle merge manually)
enriched_df.write.format("delta") \
    .mode("append") \
    .saveAsTable("catalog.schema.enriched_events")

import dlt
from pyspark.sql.functions import col, current_timestamp

# DLT handles execution order, retries, and incremental loading

@dlt.table(
    comment="Raw events ingested from JSON source"
)
def raw_events():
    return spark.read.format("json").load("/data/raw/events/")

@dlt.table(
    comment="Cleaned events with quality rules"
)
@dlt.expect_or_drop("valid_event_id", "event_id IS NOT NULL")
def cleaned_events():
    return dlt.read("raw_events").dropDuplicates(["event_id"])

@dlt.table(
    comment="Enriched events ready for analytics"
)
def enriched_events():
    return dlt.read("cleaned_events").withColumn(
        "processed_at", current_timestamp()
    )

import dlt
from pyspark.sql.functions import col, sum, avg, count

# --- Materialized Table (persisted) ---
@dlt.table(
    name="bronze_orders",
    comment="Raw orders ingested from source system",
    table_properties={
        "quality": "bronze",
        "pipelines.autoOptimize.managed": "true"
    }
)
def bronze_orders():
    return (
        spark.readStream.format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.inferColumnTypes", "true")
        .load("/data/raw/orders/")
    )

# --- View (not persisted, intermediate step) ---
@dlt.view(
    name="validated_orders",
    comment="Orders that pass validation (not persisted)"
)
def validated_orders():
    return (
        dlt.read("bronze_orders")
        .filter(col("order_total") > 0)
        .filter(col("customer_id").isNotNull())
    )

# --- Silver table using the view ---
@dlt.table(
    name="silver_orders",
    comment="Cleaned, validated orders"
)
def silver_orders():
    return dlt.read("validated_orders")

# --- Gold aggregation table ---
@dlt.table(
    name="gold_daily_revenue",
    comment="Daily revenue aggregation"
)
def gold_daily_revenue():
    return (
        dlt.read("silver_orders")
        .groupBy("order_date")
        .agg(
            sum("order_total").alias("total_revenue"),
            count("order_id").alias("order_count"),
            avg("order_total").alias("avg_order_value")
        )
    )

-- Create a DLT table using SQL
CREATE OR REFRESH STREAMING LIVE TABLE bronze_customers
COMMENT "Raw customers from CSV"
AS SELECT *
FROM cloud_files("/data/raw/customers/", "csv",
    map("header", "true", "inferSchema", "true"));

-- Create a live view
CREATE LIVE VIEW valid_customers
AS SELECT *
FROM LIVE.bronze_customers
WHERE customer_id IS NOT NULL
  AND email IS NOT NULL;

-- Create a silver table from the view
CREATE OR REFRESH LIVE TABLE silver_customers
COMMENT "Cleaned customer records"
AS SELECT * FROM LIVE.valid_customers;

import dlt
from pyspark.sql.functions import col

# Single expectation: warn but keep the row
@dlt.table()
@dlt.expect("valid_amount", "amount > 0")
def orders_warn():
    return dlt.read("raw_orders")

# Single expectation: drop failing rows
@dlt.table()
@dlt.expect_or_drop("valid_email", "email IS NOT NULL AND email LIKE '%@%'")
def customers_filtered():
    return dlt.read("raw_customers")

# Single expectation: halt pipeline on failure
@dlt.table()
@dlt.expect_or_fail("unique_transaction", "txn_id IS NOT NULL")
def transactions_strict():
    return dlt.read("raw_transactions")

# Multiple expectations at once
rules = {
    "valid_id": "order_id IS NOT NULL",
    "valid_amount": "amount > 0",
    "valid_date": "order_date >= '2020-01-01'",
    "valid_status": "status IN ('pending', 'completed', 'cancelled')"
}

@dlt.table()
@dlt.expect_all_or_drop(rules)
def validated_orders():
    return dlt.read("raw_orders")

import dlt

# --- Streaming Table: for continuously arriving data ---
@dlt.table(comment="Streaming ingestion from cloud storage")
def streaming_events():
    return (
        spark.readStream.format("cloudFiles")
        .option("cloudFiles.format", "json")
        .load("/data/events/")
    )

# --- Materialized View: aggregation that recomputes ---
@dlt.table(comment="Hourly event counts, recomputed each run")
def hourly_event_counts():
    return (
        dlt.read("streaming_events")
        .groupBy("event_type", "hour")
        .count()
    )

# --- Streaming table reading from another streaming table ---
@dlt.table(comment="Filtered stream of error events")
def error_events():
    return (
        dlt.read_stream("streaming_events")
        .filter("event_type = 'error'")
    )

{
  "name": "ecommerce_pipeline",
  "target": "ecommerce_analytics",
  "catalog": "production",
  "libraries": [
    { "notebook": { "path": "/Repos/team/pipelines/bronze" } },
    { "notebook": { "path": "/Repos/team/pipelines/silver" } },
    { "notebook": { "path": "/Repos/team/pipelines/gold" } }
  ],
  "configuration": {
    "pipelines.trigger.interval": "1 hour",
    "source_env": "production"
  },
  "clusters": [
    {
      "label": "default",
      "autoscale": {
        "min_workers": 1,
        "max_workers": 5,
        "mode": "ENHANCED"
      }
    }
  ],
  "continuous": false,
  "development": false,
  "photon": true,
  "channel": "CURRENT"
}

-- Query the DLT event log for pipeline metrics
SELECT
    timestamp,
    details:flow_definition.output_dataset AS dataset,
    details:flow_progress.metrics.num_output_rows AS rows_written,
    details:flow_progress.data_quality.expectations
FROM event_log(TABLE(ecommerce_analytics.bronze_orders))
WHERE event_type = 'flow_progress'
ORDER BY timestamp DESC
LIMIT 20;

-- Check data quality metrics
SELECT
    details:flow_definition.output_dataset AS dataset,
    details:flow_progress.data_quality.dropped_records AS dropped,
    details:flow_progress.data_quality.expectations
FROM event_log(TABLE(ecommerce_analytics.cleaned_orders))
WHERE event_type = 'flow_progress';