Databricks 4: Ingestion, Declarative Pipelines, Jobs and Streaming

This module covers how data gets into Databricks and how pipelines are built and orchestrated. Expect scenario questions such as “ingest 50,000 JSON files per hour with evolving schemas”, “implement SCD2 from a CDC feed” or “how do you rerun only the failed part of a job?”.

Naming note: Delta Live Tables (DLT) was renamed Lakeflow Declarative Pipelines, and its engine was open-sourced into Apache Spark as Spark Declarative Pipelines; Workflows became Lakeflow Jobs; managed connectors are Lakeflow Connect. Interviewers use old and new names interchangeably.

1. File ingestion: Auto Loader vs COPY INTO

Auto Loader (cloudFiles source) incrementally ingests new files from cloud storage as a stream:

(spark.readStream.format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", "/Volumes/prod/raw/_schemas/orders")   # inferred schema + evolution history
    .option("cloudFiles.schemaEvolutionMode", "addNewColumns")                  # or rescue / failOnNewColumns / none
    .option("cloudFiles.inferColumnTypes", "true")
    .load("s3://landing/orders/")
  .writeStream
    .option("checkpointLocation", "/Volumes/prod/raw/_checkpoints/orders")
    .trigger(availableNow=True)                                                 # process the backlog, then stop
    .toTable("prod.bronze.orders"))

• File discovery modes:
  • Directory listing is simple, with no extra cloud setup, and efficient for most sources with incremental listing.
  • File notification subscribes to cloud storage events (SNS/SQS, Event Grid, Pub/Sub) and scales to very high file counts without listing. Newer runtimes also offer managed file events via external locations.
• Exactly-once file tracking: processed files are tracked in the checkpoint (RocksDB), so each file is ingested once even across restarts.
• Schema handling: inference with sampling, a stored schema history, evolution modes, and the rescued data column (_rescued_data), which captures fields that don’t match the schema instead of dropping them.
• Triggers: availableNow=True gives batch-like scheduled runs with streaming bookkeeping; processingTime for continuous micro-batches.

COPY INTO is an idempotent SQL command that loads files not loaded before. It’s simple for thousands of files and SQL-first teams, but less scalable than Auto Loader for millions of files and continuous ingestion.

Lakeflow Connect provides managed connectors for SaaS apps (Salesforce, Workday, ServiceNow…) and databases (SQL Server, Postgres, etc., via CDC), writing into streaming tables with incremental sync, so there’s no custom ingestion code for common sources.

2. Lakeflow Declarative Pipelines (formerly DLT)

You declare what tables should contain; the framework handles dependency ordering, incremental processing, retries, checkpoints, infrastructure and data quality.

import dlt                      # newer API: from pyspark import pipelines as dp  (@dp.table, @dp.materialized_view)
from pyspark.sql import functions as F

@dlt.table(comment="Raw orders from Auto Loader")
def bronze_orders():
    return (spark.readStream.format("cloudFiles")
            .option("cloudFiles.format", "json").load("/Volumes/prod/landing/orders"))

@dlt.table
@dlt.expect_or_drop("valid_amount", "amount >= 0")
@dlt.expect_or_fail("has_key", "order_id IS NOT NULL")
@dlt.expect("recent", "order_ts > '2020-01-01'")            # warn only: recorded in metrics
def silver_orders():
    return (dlt.read_stream("bronze_orders")
            .withColumn("order_date", F.to_date("order_ts")))

@dlt.table                                                     # batch read → materialized view semantics
def gold_daily_revenue():
    return (dlt.read("silver_orders").groupBy("order_date")
            .agg(F.sum("amount").alias("revenue")))

Dataset types:

• Streaming table: append-only incremental processing; each input record is processed once. Use it for ingestion and incremental transformations.
• Materialized view: the result of a query, kept up to date; the engine recomputes incrementally when possible (on serverless, via Enzyme incremental refresh) or fully otherwise. Use it for aggregations, joins and anything where inputs can change.
• View: temporary, not persisted.

Expectations (data quality): expect (warn: keep rows, record metrics), expect_or_drop (drop violating rows), expect_or_fail (stop the update). Metrics land in the pipeline event log, a queryable Delta table, for quality dashboards.

CDC and SCD with AUTO CDC (formerly APPLY CHANGES INTO):

dlt.create_streaming_table("silver_customers")

dlt.create_auto_cdc_flow(                  # older name: dlt.apply_changes(...)
    target="silver_customers",
    source="bronze_customer_changes",
    keys=["customer_id"],
    sequence_by=F.col("lsn"),              # ordering column: handles out-of-order events
    apply_as_deletes=F.expr("op = 'D'"),
    except_column_list=["op", "lsn"],
    stored_as_scd_type=2,                  # 1 = overwrite in place, 2 = keep history (__START_AT/__END_AT)
)

It handles out-of-order records by sequence_by, deletes, and SCD1/SCD2 bookkeeping, replacing hundreds of lines of hand-written MERGE logic.

Pipeline settings to know:

• Triggered (run then stop) vs continuous (always on).
• Development mode (reuses the cluster, no retries) vs production mode (retries, restarts on failure).
• Serverless or classic compute, with enhanced autoscaling.
• Full refresh (recompute everything; dangerous for streaming tables whose sources no longer hold old data) vs normal incremental updates.
• Publishing to Unity Catalog catalogs/schemas.

When not to use Declarative Pipelines: very custom low-level Spark logic or tuning, complex arbitrary stateful processing, or teams standardised on another orchestration/transformation framework (e.g. dbt + Jobs).

3. Lakeflow Jobs (orchestration)

A job is a DAG of tasks: notebook, Python script/wheel, SQL (queries, dashboards, alerts), dbt, pipeline, JAR, Spark submit, “run job” (call another job), and control-flow tasks.

Interview points:

• Use table-update or file-arrival triggers instead of “run at 4am and hope upstream finished”.
• Repair run plus idempotent tasks is the efficient recovery story.
• Jobs are defined as code with Databricks Asset Bundles (YAML) and deployed via CI/CD (see module 5).
• External orchestrators (Airflow, Dagster) can trigger Databricks jobs via operators/APIs when a company standardises on them; a common pattern is Airflow for cross-system orchestration and Databricks Jobs for Databricks-internal DAGs.

4. Structured Streaming on Databricks

• Checkpoints store source offsets, state and sink commit info. Each streaming query needs its own checkpoint location (in a volume or cloud path), and changing the query in incompatible ways (e.g. a different aggregation) requires a new checkpoint.
• Triggers: availableNow (incremental batch: process everything new, then stop, which is cost-efficient for hourly SLAs), processingTime="1 minute", continuous pipelines for low latency. Newer runtimes add a real-time mode for very low latency workloads.
• State store: use RocksDB for large state (spark.sql.streaming.stateStore.providerClass), plus changelog checkpointing for faster checkpoints.
• Watermarks bound state for aggregations, stream-stream joins and dedup (dropDuplicatesWithinWatermark).
• Rate limiting: maxFilesPerTrigger / maxBytesPerTrigger (Auto Loader), maxOffsetsPerTrigger (Kafka) to control batch size and backpressure.
• Delta as source: stream from a Delta table (appends), or read its change data feed (readChangeFeed) to also get updates and deletes; skipChangeCommits to ignore non-append commits.
• Exactly-once to Delta: the Delta sink records the batch id in the transaction log, so replays don’t duplicate; foreachBatch + MERGE should use txnAppId/txnVersion or idempotent logic.
• Monitoring: StreamingQueryListener, the query progress metrics (input rate, processing rate, batch duration, state size), and the pipeline event log for declarative pipelines.

Interview questions