Auto Loader and Structured Streaming on Azure Databricks: Checkpoints, Schema Evolution, and Exactly-Once

Almost every Azure Databricks project I've seen starts the same way: files land in ADLS Gen2, and something has to pick up only the new ones, reliably, without reprocessing yesterday's data. Auto Loader plus Structured Streaming is the standard answer, and it's also a favourite interview topic because the details (checkpoints, schema evolution, exactly-once) separate people who've run it in production from people who've only read the docs.

Here's a practical walkthrough with PySpark code you can drop into a notebook.

1. The minimal Auto Loader pipeline

from pyspark.sql import functions as F

source_path = "abfss://[e-mail]/orders/"
checkpoint  = "abfss://[e-mail]/bronze_orders/"

raw = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", checkpoint + "_schema")
    .option("cloudFiles.inferColumnTypes", "true")
    .load(source_path))

bronze = (raw
    .withColumn("_ingested_at", F.current_timestamp())
    .withColumn("_source_file", F.col("_metadata.file_path")))

(bronze.writeStream
    .option("checkpointLocation", checkpoint)
    .trigger(availableNow=True)
    .toTable("bronze.orders"))

A few things worth being able to explain:

• cloudFiles is Auto Loader. It tracks which files it has already ingested (in RocksDB inside the checkpoint), so each file is processed exactly once, even if you have millions of them.
• schemaLocation is where Auto Loader stores the inferred schema and its history. Without it you'd have to pass an explicit schema.
• inferColumnTypes makes JSON/CSV columns come out as real types instead of all strings.
• _metadata.file_path gives you lineage back to the source file for free. Very handy when someone asks "which file did this bad row come from?"

2. availableNow vs. continuous streaming

The trigger is a cost decision, not just a technical one.

# Process everything that's new, in multiple micro-batches, then stop
.trigger(availableNow=True)

# Keep running, start a micro-batch every 5 minutes
.trigger(processingTime="5 minutes")

availableNow=True is what I'd use for most batch-style ingestion: schedule the notebook as a Databricks job every hour, and the cluster only lives as long as there's work. It replaced the older trigger(once=True), which squeezed everything into a single batch and could blow up memory after a long backlog. availableNow respects rate limits, so you can cap each batch:

.option("cloudFiles.maxFilesPerTrigger", 1000)
.option("cloudFiles.maxBytesPerTrigger", "10g")

When both are set, a batch stops at whichever limit is hit first.

Interview answer in one line: same code, same checkpoint, so you can move from an hourly availableNow job to an always-on stream later without reprocessing anything.

3. Checkpoints: the rules people break

The checkpoint is the stream's memory: source offsets, which batches were committed, any state, and for Auto Loader the list of discovered files. Rules I'd want any teammate to know:

1. One checkpoint per stream. Two queries sharing a checkpoint will corrupt each other's progress.
2. Don't delete it to "fix" a problem unless you actually want to reprocess everything from the start of the source.
3. Treat it as tied to the query. Changing the source, or making big changes to aggregations/joins with state, can make an old checkpoint incompatible. Adding a filter or a new projected column is usually fine.
4. Keep it next to the data's lifecycle. If you drop and recreate the target table, you almost always want a new checkpoint too.

To see what Auto Loader has actually ingested, Databricks has a table-valued function you can query directly:

SELECT path, commit_time
FROM cloud_files_state('abfss://[e-mail]/bronze_orders/')
ORDER BY commit_time DESC
LIMIT 20;

4. Schema evolution without 2 a.m. pages

Upstream teams add columns. The question is what your pipeline does when they do.

.option("cloudFiles.schemaEvolutionMode", "addNewColumns")  # default when you don't supply a schema

The modes:

Mode	What happens on a new column
addNewColumns	Stream fails once, schema is updated in schemaLocation, next run includes the column
rescue	Schema never changes; unexpected data goes into _rescued_data
failOnNewColumns	Stream fails and stays failed until you change the schema
none	New columns are ignored

The "fails once" behaviour of addNewColumns surprises people. It's by design: the stream stops, records the new schema, and a restart picks it up. In a Databricks job, set the task to retry on failure and it heals itself. On the write side, add .option("mergeSchema", "true") so the Delta table accepts the new column too.

I also like pinning the columns that matter with schema hints, so inference can't guess wrong:

.option("cloudFiles.schemaHints", "order_id BIGINT, amount DECIMAL(18,2), order_ts TIMESTAMP")

And always check _rescued_data. Auto Loader adds it by default when it infers the schema, and it captures values that didn't fit (a string in a numeric column, a field with different casing). A simple data quality check:

bad = spark.table("bronze.orders").filter("_rescued_data IS NOT NULL")
display(bad.select("_source_file", "_rescued_data").limit(50))

5. Exactly-once with foreachBatch

Writing straight to a Delta table with toTable is exactly-once out of the box: Delta and the checkpoint together make sure a batch that's retried doesn't get committed twice.

foreachBatch is where people lose that guarantee. It gives you full batch APIs (MERGE, writing to several tables), but if the job crashes after your write and before the checkpoint commit, the same batch runs again. Two ways to stay safe:

Option A: make the write idempotent with Delta's transaction options.

APP_ID = "bronze_to_silver_orders"  # stable, unique per stream

def write_batch(batch_df, batch_id):
    (batch_df.write
        .format("delta")
        .mode("append")
        .option("txnAppId", APP_ID)
        .option("txnVersion", batch_id)
        .saveAsTable("silver.orders_append"))

(spark.readStream.table("bronze.orders")
    .writeStream
    .foreachBatch(write_batch)
    .option("checkpointLocation", ".../checkpoints/silver_orders/")
    .trigger(availableNow=True)
    .start())

Delta remembers the highest txnVersion it committed for that txnAppId and silently skips a repeat. If you write to two tables in one batch, use the same pair of options on each write.

Option B: use MERGE and deduplicate inside the batch.

from delta.tables import DeltaTable
from pyspark.sql import Window

def upsert_batch(batch_df, batch_id):
    w = Window.partitionBy("order_id").orderBy(F.col("order_ts").desc())
    latest = (batch_df
        .withColumn("rn", F.row_number().over(w))
        .filter("rn = 1")
        .drop("rn"))

    (DeltaTable.forName(batch_df.sparkSession, "silver.orders").alias("t")
        .merge(latest.alias("s"), "t.order_id = s.order_id")
        .whenMatchedUpdateAll(condition="s.order_ts > t.order_ts")
        .whenNotMatchedInsertAll()
        .execute())

Replaying this batch produces the same end state, so retries are harmless. The in-batch dedup matters because MERGE fails if several source rows match the same target row.

Note batch_df.sparkSession instead of the global spark: on newer runtimes and shared clusters that's the safe way to reach the session inside the function.

6. Directory listing vs. file notifications

By default Auto Loader discovers files by listing the input directory. That's fine for most workloads, and incremental listing keeps it cheap when file names are lexically ordered (like date-based paths).

For very high file volumes, file notification mode subscribes to storage events instead (Event Grid and a storage queue on Azure):

.option("cloudFiles.useNotifications", "true")

It needs permissions to create or use those Azure resources, so it's a conversation with your platform team, and newer Databricks workspaces also offer managed file events through Unity Catalog external locations. In an interview, knowing why you'd switch (listing cost and latency growing with directory size) matters more than the exact setup.

7. Monitoring a running stream

q = spark.streams.active[0]
print(q.status)
print(q.lastProgress["numInputRows"], q.lastProgress["batchDuration"])

For Auto Loader, lastProgress also includes a source metric for the backlog (numFilesOutstanding / numBytesOutstanding), which is the number I'd alert on: a growing backlog means ingestion isn't keeping up.

Quick checklist

• cloudFiles + schemaLocation + a unique checkpoint per stream.
• availableNow=True for scheduled jobs, with maxFilesPerTrigger/maxBytesPerTrigger caps.
• addNewColumns plus job retries, mergeSchema on write, schema hints for key columns.
• Watch _rescued_data.
• In foreachBatch, use txnAppId/txnVersion or an idempotent MERGE.
• Alert on backlog, not just job failure.

---

If you're preparing for a Databricks or PySpark interview, there's a free 10-question sample with worked answers in this style. The full Azure Databricks & PySpark Interview Pack has the rest if it helps.