Write Incremental Loads with Auto Loader

Imagine your data lake is a firehose — files land every second, and you need to process only the new ones, not re-scan the entire bucket every time. If you've tried spark.read.format("json").load("/path"), you know the pain: it reads everything, and as your data grows, your pipelines slow to a crawl, costing you time and credits. In this lesson, you'll master Auto Loader — Databricks' incremental data ingestion engine — and learn to write incremental loads that automatically discover and process only the files that have arrived since your last run. By the end, you'll be able to transform a brittle batch job into a streaming-ready, cost-efficient pipeline.

The problem this lesson solves

Most data pipelines start simple: read a folder, transform, write a table. But in production, data doesn't respect your schedule. New files arrive continuously, and your pipeline either re-reads the entire dataset (wasteful and slow) or you rely on fragile bookkeeping like timestamp filters (which break across different systems).

Consider this: a typical e-commerce platform ingests thousands of JSON files per hour. If your daily batch job reads 10 GB of data every hour to find the few megabytes of new files, you're over-provisioning clusters and burning through Databricks DBUs. More critically, you're adding latency and cost to every downstream analytics decision.

Without a proper incremental strategy, you also risk duplicated or missed records — the classic 'at-least-once' headache. Auto Loader solves this by treating your cloud storage (S3, ADLS, GCS) as a streaming source that emits only the files that are new, using a built-in checkpoint mechanism.

Core concept / mental model

Think of Auto Loader as a smart file sniffer that keeps a ledger of what it has already seen. Instead of reading every file in a directory each time, it scans the metadata (often via cloud notifications) and identifies new files since the last checkpoint. It then loads only those files into your DataFrame.

Here's a mental model: imagine you're a librarian who adds new books to a reading list. You don't re-read every book in the library each day — you check the 'new arrivals' shelf. Auto Loader's checkpoint is that shelf, and it remembers the exact position.

Key terms to understand:

• Cloud file notifications: Auto Loader can subscribe to storage event notifications (e.g., S3 bucket notifications) to get near-real-time file alerts — low latency, minimal scanning.
• Directory listing: If notifications aren't available (e.g., in a sandbox), Auto Loader falls back to periodically listing the directory and comparing against the checkpoint — simpler but slightly slower.
• Checkpoint location: A directory (usually _checkpoints) where Auto Loader stores the metadata of already-processed files. It's the heart of incremental load.

How it works step by step

Here’s the logical flow you’ll follow every time you set up an incremental load with Auto Loader:

1. Define the source path: Specify the blob or directory that contains your incoming files (e.g., s3://bucket/landing_zone).
2. Choose the file format: Auto Loader supports JSON, Parquet, CSV, text, and more. You set the format option.
3. Set the checkpoint location: Specify where Auto Loader should store its state (e.g., dbfs:/mnt/checkpoints/). This is non-negotiable — without it, you can't track progress.
4. Configure file discovery mode: Decide between notifications (fast, uses cloud events) or directoryListing (fallback).
5. Read as a stream: spark.readStream.format("cloudFiles") creates a streaming DataFrame that emits new files.
6. Write incrementally: Use trigger(availableNow=True) for a batch-style incremental load, or a trigger(processingTime=...) for streaming mode, then write to a Delta table.

The magic: Auto Loader tracks which files were processed, so your query only sees new data at the next run.

Hands-on walkthrough

Let's build a complete example. Assume you have JSON files landing in dbfs:/mnt/raw_events/, and you want to maintain an incremental Delta table.

First, set up your streaming read:

from pyspark.sql.functions import col, current_timestamp

# 1. Define source and checkpoint paths
source_path = "/mnt/raw_events"
checkpoint_path = "/mnt/checkpoints/events_incremental"

events_df = (
    spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.useNotifications", "false")  # Use directoryListing for simplicity
    .option("pathGlobFilter", "*.json")
    .schema("id INT, event_type STRING, amount DOUBLE")
    .load(source_path)
    .withColumn("ingest_time", current_timestamp())
)

Next, write the stream to a Delta table. Using availableNow lets you run incremental loads in a batch-style mode — perfect for scheduled jobs:

(
    events_df.writeStream
    .format("delta")
    .options(
        checkpointLocation=checkpoint_path,
        mergeSchema="true"
    )
    .outputMode("append")
    .trigger(availableNow=True)
    .table("events_raw")
)

Run this once, and you'll see output like:

Streaming query made progress:
{
  "numInputRows": 1000,
  "processedRowsPerSecond": 50000,
  "inputRowsPerSecond": 80000
}

Now, if you drop two new JSON files into the source path and run the same cell again, only those new files are read — the checkpoint handles the rest. You'll see numInputRows equals just the count of new rows, not the total.

For true streaming (near-real-time), switch the trigger:

(
    events_df.writeStream
    .trigger(processingTime="10 seconds")
    .start()
)

This runs continuously and picks up new files as they arrive — ideal for live dashboards.

Compare options / when to choose what

Auto Loader isn't your only option for incremental loads. Let's compare it with common alternatives:

When to choose Auto Loader: if you need at-least-once processing, schema evolution, and you're already on Databricks. Use COPY INTO if your only task is to load files to a table and you don't need a DataFrame for complex transformations. Avoid timestamp filters for production—they'll fail when files arrive late or zones shift.

Troubleshooting & edge cases

1. "Unable to list files" or permission errors

• Check your service principal has list and read permissions on the source directory. Auto Loader needs both.
• For notifications, ensure the storage event subscriptions are configured and that Databricks has the right IAM to subscribe (e.g., s3:PutBucketNotification).

2. Files are processed twice

• This happens if you use a shared checkpoint across multiple streams. Never reuse the same checkpointPath for different source paths.
• Also, if you manually move files into the source folder after the folder was already scanned, Auto Loader might re-process them. Keep your landing zone append-only.

3. Inferred schema changes break your load

• Auto Loader can infer schema, but if a new file has an extra column, you'll get an error. Use schemaHints or provide an explicit schema at creation.
• Enable cloudFiles.schemaEvolutionMode (e.g., "addNewColumns") when you expect evolving data.

.option("cloudFiles.schemaEvolutionMode", "addNewColumns")

4. Stream stops silently in notebooks

• If you don't use awaitTermination() or a query that writes to a Delta table, the stream may not start. Always attach the stream to a sink before ending the cell.
• When using availableNow, the query finishes automatically — you don't need awaitTermination.

5. High latency with directoryListing mode

• If you set "cloudFiles.useNotifications": "false", Auto Loader polls the directory—this can add minutes of latency. For production, enable notifications or benchmark your polling interval.

What you learned & what's next

You now understand how to write incremental loads with Auto Loader — from setting up a streaming DataFrame to managing checkpoints and troubleshooting common issues. You can explain the core concept (file discovery + checkpoint tracking), and you've done a hands-on exercise with both availableNow and continuous triggers.

These skills directly prepare you for the next lesson in this path: Automating production ETL pipelines (scheduling, monitoring, and alerts). You'll apply the same incremental patterns but add orchestration and reliability layers.

Keep your checkpoints separate, your source directories append-only, and your schema explicit — and you'll build robust, cost-efficient pipelines on Databricks.