Advanced PySpark: DLT, Auto Loader & Optimization

# List files in cloud storage (ADLS Gen2 / S3 / GCS)
files = dbutils.fs.ls("abfss://bronze@datalake.dfs.core.windows.net/raw_events/")
for f in files:
    print(f.name, f.size, f.modificationTime)

# Move a file (staging → processed)
dbutils.fs.mv(
    "abfss://landing@datalake.dfs.core.windows.net/orders/file.json",
    "abfss://processed@datalake.dfs.core.windows.net/orders/file.json"
)

# Copy a directory recursively
dbutils.fs.cp(
    "abfss://bronze@datalake.dfs.core.windows.net/snapshot/",
    "abfss://backup@datalake.dfs.core.windows.net/snapshot_2026_05_07/",
    recurse=True
)

# Delete old files (recurse=True for directories)
dbutils.fs.rm("abfss://landing@datalake.dfs.core.windows.net/orders/old_batch/", recurse=True)

# Mount cloud storage (classic workspaces without Unity Catalog)
dbutils.fs.mount(
    source="wasbs://container@storageaccount.blob.core.windows.net",
    mount_point="/mnt/raw",
    extra_configs={
        "fs.azure.account.key.storageaccount.blob.core.windows.net":
            dbutils.secrets.get(scope="azure-kv", key="storage-account-key")
    }
)

# Read a secret — the value is never printed in notebook output
jdbc_password   = dbutils.secrets.get(scope="prod-secrets", key="sql-db-password")
kafka_api_key   = dbutils.secrets.get(scope="prod-secrets", key="confluent-api-key")
openai_api_key  = dbutils.secrets.get(scope="prod-secrets", key="openai-api-key")

# List all keys in a scope (values are hidden)
dbutils.secrets.list("prod-secrets")
# Returns: [SecretMetadata(key='sql-db-password'), SecretMetadata(key='confluent-api-key'), ...]

# Use in JDBC connection
df = (
    spark.read
    .format("jdbc")
    .option("url",      "jdbc:sqlserver://sqlserver.database.windows.net:1433;database=OrdersDB")
    .option("dbtable",  "dbo.Orders")
    .option("user",     "etl_user")
    .option("password", jdbc_password)
    .option("driver",   "com.microsoft.sqlserver.jdbc.SQLServerDriver")
    .load()
)

# Declare widgets at the top of the notebook
dbutils.widgets.text(
    name="start_date",
    defaultValue="2026-05-01",
    label="Start Date (yyyy-MM-dd)"
)
dbutils.widgets.text(
    name="end_date",
    defaultValue="2026-05-07",
    label="End Date (yyyy-MM-dd)"
)
dbutils.widgets.dropdown(
    name="environment",
    defaultValue="dev",
    choices=["dev", "staging", "prod"],
    label="Target Environment"
)
dbutils.widgets.combobox(
    name="table_suffix",
    defaultValue="orders",
    choices=["orders", "events", "customers"],
    label="Table Name Suffix"
)

# Read widget values anywhere in the notebook
start_date  = dbutils.widgets.get("start_date")
end_date    = dbutils.widgets.get("end_date")
environment = dbutils.widgets.get("environment")

catalog     = "prod" if environment == "prod" else "dev"
table_name  = f"{catalog}.silver.{dbutils.widgets.get('table_suffix')}"

print(f"Processing {table_name} from {start_date} to {end_date}")

# Filter data using the widget parameters
df = spark.sql(f"""
    SELECT * FROM {table_name}
    WHERE created_at BETWEEN '{start_date}' AND '{end_date}'
""")

# Run a child notebook and pass parameters to it
result = dbutils.notebook.run(
    path="/Pipelines/ingest_bronze",
    timeout_seconds=3600,
    arguments={
        "start_date":  start_date,
        "end_date":    end_date,
        "environment": environment,
    }
)
print(f"Child notebook returned: {result}")

# Child notebook signals completion by calling:
dbutils.notebook.exit("SUCCESS: 12,450 rows ingested")

# For error propagation:
try:
    dbutils.notebook.run("/Pipelines/risky_job", timeout_seconds=600, arguments={})
except Exception as e:
    print(f"Child notebook failed: {e}")
    raise   # re-raise to fail the parent job

import dlt
from pyspark.sql.functions import col, current_timestamp, to_date, sha2, concat_ws
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType

# ─── BRONZE LAYER ────────────────────────────────────────────────────────────

@dlt.table(
    name="raw_orders",
    comment="Raw order events ingested from cloud storage via Auto Loader",
    table_properties={
        "delta.autoOptimize.optimizeWrite": "true",
        "delta.autoOptimize.autoCompact":   "true",
    }
)
def raw_orders():
    """
    Auto Loader: incrementally ingest new JSON files from ADLS Gen2.
    DLT manages the checkpoint automatically.
    """
    return (
        spark.readStream
        .format("cloudFiles")                    # Auto Loader format
        .option("cloudFiles.format", "json")
        .option("cloudFiles.schemaLocation",
                "abfss://checkpoints@lake.dfs.core.windows.net/dlt/raw_orders/schema")
        .option("cloudFiles.inferColumnTypes", "true")
        .load("abfss://landing@lake.dfs.core.windows.net/orders/")
        .withColumn("_ingested_at", current_timestamp())
        .withColumn("_source_file", col("_metadata.file_path"))
    )

# ─── SILVER LAYER ────────────────────────────────────────────────────────────

@dlt.table(
    name="clean_orders",
    comment="Validated and deduplicated orders — Silver layer"
)
@dlt.expect("order_id is not null",    "order_id IS NOT NULL")
@dlt.expect("customer_id is not null", "customer_id IS NOT NULL")
@dlt.expect("valid status",
    "status IN ('pending', 'processing', 'shipped', 'completed', 'cancelled')")
@dlt.expect_or_drop("total is positive", "total > 0")     # drop rows that fail
@dlt.expect_or_fail("no duplicate order_id",              # fail the pipeline if violated
    "COUNT(order_id) OVER (PARTITION BY order_id) = 1")
def clean_orders():
    """
    Clean raw orders: type-cast, validate, deduplicate.
    Expectations are evaluated and metrics surfaced in the DLT pipeline UI.
    """
    return (
        dlt.read_stream("raw_orders")              # reference upstream DLT table
        .dropDuplicates(["order_id"])
        .withColumn("total",      col("total").cast(DoubleType()))
        .withColumn("created_at", col("created_at").cast(TimestampType()))
        .withColumn("order_date", to_date(col("created_at")))
        .withColumn("row_hash",
            sha2(concat_ws("|", col("order_id"), col("total"), col("status")), 256))
    )


# Expectation action reference:
# @dlt.expect(name, condition)           → warn in metrics, keep bad rows
# @dlt.expect_or_drop(name, condition)   → silently drop rows that fail
# @dlt.expect_or_fail(name, condition)   → halt the pipeline on violation

# ─── GOLD LAYER ──────────────────────────────────────────────────────────────

@dlt.table(
    name="monthly_revenue",
    comment="Monthly revenue summary — Gold layer, refreshed on each pipeline run"
)
def monthly_revenue():
    return (
        dlt.read("clean_orders")                   # batch read, not streaming
        .filter(col("status") == "completed")
        .groupBy(
            col("order_date").substr(1, 7).alias("month")   # "2026-05"
        )
        .agg(
            {"total": "sum", "order_id": "count"}
        )
        .withColumnRenamed("sum(total)",     "revenue")
        .withColumnRenamed("count(order_id)", "order_count")
    )

# Schema hint: tell Auto Loader the expected schema (avoids full schema inference on every restart)
order_schema = StructType([
    StructField("order_id",    StringType(),  nullable=False),
    StructField("customer_id", StringType(),  nullable=False),
    StructField("total",       DoubleType(),  nullable=True),
    StructField("status",      StringType(),  nullable=True),
    StructField("created_at",  TimestampType(), nullable=True),
])

stream_df = (
    spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    # Schema location: Auto Loader stores inferred schema here for schema evolution tracking
    .option("cloudFiles.schemaLocation",
            "abfss://checkpoints@lake.dfs.core.windows.net/autoloader/orders/schema")
    # File notification mode: Azure Event Grid / SQS (faster than directory listing)
    .option("cloudFiles.useNotifications", "true")
    .option("cloudFiles.subscriptionId",  dbutils.secrets.get("az-secrets", "subscription-id"))
    .option("cloudFiles.tenantId",        dbutils.secrets.get("az-secrets", "tenant-id"))
    .option("cloudFiles.clientId",        dbutils.secrets.get("az-secrets", "sp-client-id"))
    .option("cloudFiles.clientSecret",    dbutils.secrets.get("az-secrets", "sp-client-secret"))
    # If schema evolves (new columns added), merge rather than fail
    .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
    .schema(order_schema)
    .load("abfss://landing@lake.dfs.core.windows.net/orders/")
)

# Write to Bronze Delta table — checkpoint ensures exactly-once processing
query = (
    stream_df
    .withColumn("_ingested_at", current_timestamp())
    .writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation",
            "abfss://checkpoints@lake.dfs.core.windows.net/autoloader/orders/checkpoint")
    .trigger(availableNow=True)     # process all backlog then stop (batch-like behavior)
    .toTable("catalog.bronze.raw_orders")
)

query.awaitTermination()

# Process all available files then stop (scheduled batch pattern)
.trigger(availableNow=True)

# Run continuously, process new files every 30 seconds
.trigger(processingTime="30 seconds")

# Run once and stop (deprecated — use availableNow instead)
.trigger(once=True)

import dlt
from pyspark.sql.functions import from_json, col
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType

ORDER_EVENT_SCHEMA = StructType([
    StructField("order_id",    StringType()),
    StructField("customer_id", StringType()),
    StructField("total",       DoubleType()),
    StructField("event_type",  StringType()),
    StructField("event_ts",    TimestampType()),
])

@dlt.table(
    name="raw_order_events_kafka",
    comment="Raw Kafka events from the orders topic — Bronze layer"
)
def raw_order_events_kafka():
    return (
        spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers",
                dbutils.secrets.get("kafka-secrets", "bootstrap-servers"))
        .option("kafka.security.protocol", "SASL_SSL")
        .option("kafka.sasl.mechanism",    "PLAIN")
        .option("kafka.sasl.jaas.config",
                f'org.apache.kafka.common.security.plain.PlainLoginModule required '
                f'username="{dbutils.secrets.get("kafka-secrets", "api-key")}" '
                f'password="{dbutils.secrets.get("kafka-secrets", "api-secret")}";')
        .option("subscribe",          "order-events")
        .option("startingOffsets",    "latest")
        .option("failOnDataLoss",     "false")
        .load()
        # Kafka value is binary — cast to string and parse JSON
        .select(
            col("offset"),
            col("partition"),
            col("timestamp").alias("kafka_ts"),
            from_json(col("value").cast("string"), ORDER_EVENT_SCHEMA).alias("data")
        )
        .select("offset", "partition", "kafka_ts", "data.*")
    )


@dlt.table(name="clean_order_events")
@dlt.expect_or_drop("valid order_id", "order_id IS NOT NULL")
@dlt.expect_or_drop("valid total",    "total > 0")
def clean_order_events():
    return dlt.read_stream("raw_order_events_kafka")

# Checkpoint best practices
query = (
    clean_df.writeStream
    .format("delta")
    .outputMode("append")
    # Store checkpoint on reliable cloud storage — NOT local disk
    .option("checkpointLocation",
            "abfss://checkpoints@lake.dfs.core.windows.net/streams/orders/v1")
    # Never change the checkpoint path after production deployment
    # If you need to change the query, create a new checkpoint path (v2, v3)
    .trigger(processingTime="60 seconds")
    .toTable("catalog.silver.orders_stream")
)

# Monitor query progress
import time
for _ in range(5):
    progress = query.lastProgress
    if progress:
        print(f"Batch: {progress['batchId']} | "
              f"Input rows: {progress['numInputRows']} | "
              f"Processing rate: {progress['processedRowsPerSecond']:.0f} rows/s")
    time.sleep(30)

# Photon is enabled at the cluster level — no code change needed
# Enable via: Cluster Configuration → Runtime → "Use Photon Acceleration"
# Or in cluster JSON:
# "spark_conf": {"spark.databricks.photon.enabled": "true"}

# Photon accelerates: aggregations, joins, sort, filter, columnar scan
# Photon does NOT accelerate: Python UDFs, RDD operations, map/flatMap
# Best practice: eliminate Python UDFs — replace with SQL functions where possible

# Instead of a Python UDF:
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

# BAD — bypasses Photon
@udf(StringType())
def clean_email_udf(email: str) -> str:
    return email.lower().strip() if email else None

# GOOD — Photon-accelerated built-in functions
from pyspark.sql.functions import lower, trim
df = df.withColumn("email", lower(trim(col("email"))))

-- Enable liquid clustering on a new table
CREATE TABLE catalog.silver.events
CLUSTER BY (customer_id, event_date)
USING DELTA;

-- Enable on an existing table (migrates incrementally on next OPTIMIZE)
ALTER TABLE catalog.silver.events
CLUSTER BY (customer_id, event_date);

-- Optimize runs the same — liquid clustering is applied automatically
OPTIMIZE catalog.silver.events;

Single Node cluster:
  Driver = Worker (same VM)
  No network shuffle between nodes
  Best for: ML training on pandas/sklearn, small ETL, notebook exploration

Multi-Node cluster:
  Driver VM + N Worker VMs
  Data distributed across workers via Shuffle
  Best for: large-scale PySpark jobs, DLT pipelines, SQL warehouses

{
  "autoscale": {
    "min_workers": 2,
    "max_workers": 20
  },
  "spark_conf": {
    "spark.databricks.adaptive.autoBroadcastJoinThreshold": "50MB",
    "spark.sql.shuffle.partitions": "auto"
  }
}

{
  "aws_attributes": {
    "availability": "SPOT_WITH_FALLBACK",
    "spot_bid_price_percent": 100,
    "first_on_demand": 1
  }
}

# Best practice for scheduled jobs:
# 1. Develop on an interactive cluster
# 2. In the Job definition, specify a "new_cluster" config (job cluster)
# 3. For time-sensitive jobs where 5 minutes startup matters, use a
#    "existing_cluster_id" to reuse a warm interactive cluster

import dlt
from pyspark.sql.functions import (
    col, current_timestamp, to_date, lower, trim,
    sha2, concat_ws, from_json, coalesce, lit
)
from pyspark.sql.types import (
    StructType, StructField, StringType, DoubleType, TimestampType
)

# ── Shared schema ─────────────────────────────────────────────────────────────

ORDER_SCHEMA = StructType([
    StructField("order_id",    StringType(),  False),
    StructField("customer_id", StringType(),  False),
    StructField("total",       DoubleType(),  True),
    StructField("status",      StringType(),  True),
    StructField("email",       StringType(),  True),
    StructField("created_at",  TimestampType(), True),
])

# ── Bronze: Auto Loader from ADLS ─────────────────────────────────────────────

@dlt.table(
    name="bronze_orders_files",
    comment="Raw order JSON files ingested incrementally via Auto Loader",
    table_properties={"delta.autoOptimize.optimizeWrite": "true"}
)
def bronze_orders_files():
    return (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format",         "json")
        .option("cloudFiles.schemaLocation",
                "abfss://checkpoints@lake.dfs.core.windows.net/dlt/schema/orders_files")
        .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
        .option("cloudFiles.inferColumnTypes",    "true")
        .schema(ORDER_SCHEMA)
        .load("abfss://landing@lake.dfs.core.windows.net/orders/")
        .withColumn("_ingested_at", current_timestamp())
        .withColumn("_source",      lit("files"))
    )

# ── Bronze: Kafka stream ───────────────────────────────────────────────────────

@dlt.table(
    name="bronze_orders_kafka",
    comment="Real-time order events consumed from Kafka"
)
def bronze_orders_kafka():
    return (
        spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers",
                dbutils.secrets.get("kafka-secrets", "bootstrap-servers"))
        .option("kafka.security.protocol", "SASL_SSL")
        .option("kafka.sasl.mechanism",    "PLAIN")
        .option("kafka.sasl.jaas.config",
                f'org.apache.kafka.common.security.plain.PlainLoginModule required '
                f'username="{dbutils.secrets.get("kafka-secrets", "key")}" '
                f'password="{dbutils.secrets.get("kafka-secrets", "secret")}";')
        .option("subscribe",       "order-events")
        .option("startingOffsets", "latest")
        .option("failOnDataLoss",  "false")
        .load()
        .select(
            from_json(col("value").cast("string"), ORDER_SCHEMA).alias("d"),
            col("timestamp").alias("_kafka_ts")
        )
        .select("d.*", "_kafka_ts")
        .withColumn("_ingested_at", current_timestamp())
        .withColumn("_source",      lit("kafka"))
    )

# ── Silver: Union both sources, validate, clean ───────────────────────────────

@dlt.table(
    name="silver_orders",
    comment="Validated and cleaned orders from all sources — Silver layer",
    table_properties={
        "delta.autoOptimize.optimizeWrite": "true",
        "delta.dataSkippingNumIndexedCols": "5",
    }
)
@dlt.expect("order_id not null",    "order_id IS NOT NULL")
@dlt.expect("customer_id not null", "customer_id IS NOT NULL")
@dlt.expect_or_drop("positive total",   "total > 0")
@dlt.expect_or_drop("valid status",
    "status IN ('pending','processing','shipped','completed','cancelled')")
def silver_orders():
    # Union the two Bronze sources
    files_stream = dlt.read_stream("bronze_orders_files")
    kafka_stream = dlt.read_stream("bronze_orders_kafka")

    combined = files_stream.unionByName(kafka_stream, allowMissingColumns=True)

    return (
        combined
        .dropDuplicates(["order_id"])
        .withColumn("email",      lower(trim(col("email"))))
        .withColumn("order_date", to_date(col("created_at")))
        .withColumn("row_hash",
            sha2(concat_ws("|", col("order_id"), col("total"), col("status")), 256))
        .withColumn("_cleaned_at", current_timestamp())
    )

# DLT pipelines are deployed via Databricks UI or the REST API / Terraform
# Terraform example:
#
# resource "databricks_pipeline" "orders_dlt" {
#   name = "orders-medallion-pipeline"
#   storage = "abfss://dlt@lake.dfs.core.windows.net/pipelines/orders"
#   cluster {
#     autoscale { min_workers = 1; max_workers = 5 }
#     node_type_id = "Standard_DS3_v2"
#   }
#   library { notebook { path = "/Pipelines/orders_dlt_pipeline" } }
#   continuous = false   # triggered mode: run on schedule
#   channel = "CURRENT"
# }