Incremental Ingestion with Auto Loader

Introduction

This lesson is the second part of the Data Ingestion and Loading section of the official exam outline (about 21% of the scored questions, roughly nine or ten of 45). Lesson 3 showed COPY INTO, which works well for thousands of files. Auto Loader is the Databricks answer when files arrive continuously, when there are very many of them, or when the schema changes over time. It is one of the most tested ingestion topics because its syntax (cloudFiles) and its options are easy to confuse.

After this lesson you will be able to write an Auto Loader stream in Python and SQL, choose between directory listing and file notification mode, explain how schemas are inferred, stored and evolved, use the rescued data column, and run Auto Loader as an incremental batch with availableNow.

Learning objectives

Explain what Auto Loader does and why it scales better than COPY INTO for large file counts.
Write the cloudFiles source with cloudFiles.format and the target table, checkpoint and schema location.
Compare directory listing mode and file notification mode.
Use cloudFiles.inferColumnTypes, cloudFiles.schemaHints and cloudFiles.schemaLocation.
Choose a cloudFiles.schemaEvolutionMode (addNewColumns, rescue, failOnNewColumns, none) and explain the rescued data column.
Explain checkpoints and the availableNow trigger for scheduled, incremental batch processing.

1. What Auto Loader is

Auto Loader is a Structured Streaming source called cloudFiles. It incrementally and efficiently discovers new files as they land in cloud storage (ADLS, S3, GCS) or in a Unity Catalog volume, and loads them into a Delta table. It can process billions of files, and it supports JSON, CSV, XML, Parquet, Avro, ORC, text and binary files.

Its main properties, which the exam keeps returning to, are:

Incremental: each file is processed exactly once. Progress is stored in a checkpoint, so a restarted stream resumes where it stopped.
Scalable: it discovers files without listing the whole directory every time (in notification mode) or with optimised incremental listing (in directory listing mode).
Schema aware: it can infer the schema, store it, and evolve it as new columns appear, while keeping unexpected data in a rescued data column.
Flexible latency: the same code can run continuously (seconds of latency) or on a schedule with the availableNow trigger (incremental batch).

Question	COPY INTO	Auto Loader
Interface	SQL command	Structured Streaming source (`cloudFiles`), also usable in streaming tables and pipelines
File tracking	Table metadata	Checkpoint (RocksDB state)
Typical scale	Thousands of files	Millions to billions of files
Schema evolution	`mergeSchema`	Schema location, evolution modes, rescued data
Latency	Batch only	Continuous or triggered

2. Basic syntax

Python

(spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", "/Volumes/prod/bronze/_schemas/events")
    .load("s3://acme-landing/events/")
  .writeStream
    .option("checkpointLocation", "/Volumes/prod/bronze/_checkpoints/events")
    .trigger(availableNow=True)
    .toTable("prod.bronze.events"))

Reading the code from top to bottom:

format("cloudFiles") selects Auto Loader. It is cloudFiles, not autoloader.
cloudFiles.format is the format of the source files (json, csv, parquet, ...). A frequent mistake is to write .format("json") and lose all Auto Loader behaviour.
cloudFiles.schemaLocation is the folder where Auto Loader stores the inferred schema and its history. It is required when the schema is inferred. It may be the same folder as the checkpoint.
load(path) points at the input directory.
checkpointLocation stores the stream progress. It must be unique per stream.
toTable(...) writes to a Unity Catalog Delta table.

SQL

In SQL, you read files incrementally with a streaming table and read_files, which uses Auto Loader under the hood:

CREATE OR REFRESH STREAMING TABLE prod.bronze.events
AS SELECT *, _metadata.file_path AS source_file
FROM STREAM read_files(
  's3://acme-landing/events/',
  format => 'json'
);

The keyword STREAM is what makes the read incremental; without it read_files is a plain batch read that rescans everything. In Lakeflow Spark Declarative Pipelines (formerly Delta Live Tables or DLT) the same pattern defines a bronze streaming table, and the pipeline manages the schema location and checkpoints for you.

Worked example: reading with the wrong format option

Scenario. A developer writes the following and notices that new files in the folder are not picked up after the first run, and no schema is stored.

df = (spark.readStream.format("json")
        .load("s3://acme-landing/events/"))

What is wrong, and how do you fix it?

Reasoning.

format("json") is the plain Structured Streaming file source. It also reads a directory, but it needs an explicit schema, it lists the whole directory on every trigger, and it has no schema evolution or rescued data.
Auto Loader is selected with format("cloudFiles") and the file format moves to cloudFiles.format.
Add cloudFiles.schemaLocation so the inferred schema is persisted.

df = (spark.readStream.format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.schemaLocation", "/Volumes/prod/bronze/_schemas/events")
        .load("s3://acme-landing/events/"))

Why other fixes are weaker. Switching to spark.read makes the load batch and non-incremental. Adding inferSchema does not give Auto Loader's file discovery or evolution. Running COPY INTO would work for modest volumes, but the question asks about the streaming source.

3. How Auto Loader discovers files

Auto Loader supports two file detection modes.

Directory listing mode (default)

Auto Loader lists the input directory and uses the listing to find new files. It needs only read access to the path, so it is the easiest to start with: no cloud permissions to create queues or subscriptions. It uses optimised incremental listing, which keeps it fast for directories that receive files in lexical order (for example date-partitioned paths like /2026/10/24/).

Choose it when:

you want a quick setup with minimal cloud permissions;
the input volume is moderate, or files arrive in lexical order;
you read from a Unity Catalog volume.

File notification mode

Auto Loader sets up cloud services (for example a queue and an event subscription) that notify it when a file arrives. Because it does not list the directory, it scales to very large directories and very high file counts at lower listing cost. Enable it with:

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

It requires additional cloud permissions to create the notification resources (or an administrator to pre-create them). Choose it when:

the input directory is huge or files arrive in a non-lexical order, making listing slow and costly;
you need the lowest discovery latency at scale;
you can obtain the permissions to create event resources.

Aspect	Directory listing	File notification
Option	default	`cloudFiles.useNotifications = true`
Cloud permissions	Read on the path	Also create queue and event subscription
Scale	Good, can become slow on very large directories	Best for very large and fast-growing directories
Setup effort	Minimal	Higher

Example. A telecom company writes 40 million call-detail files per month into a flat bucket with random-looking names. Listing the bucket on each trigger takes more than ten minutes, so jobs overlap. The right move is file notification mode, because it replaces listing with event notifications. A small start-up that drops a few hundred files per day into a dated folder should stay with directory listing, since it needs no extra permissions.

4. Schema inference and hints

When you do not provide a schema, Auto Loader samples the data (the first 50 GB or 1,000 files, whichever limit is reached first) to infer it, then stores it in the _schemas directory inside cloudFiles.schemaLocation. Later runs reuse the stored schema and evolve it when needed.

Important behaviours:

For JSON, CSV and XML, Auto Loader infers all columns as strings by default, because text formats do not carry types and sampling cannot guarantee a type. Set cloudFiles.inferColumnTypes to true to infer real types (bigint, double, timestamp, nested structs and arrays).
For Parquet, Avro and ORC, the types are already in the files, so they are used as they are.
cloudFiles.schemaHints overrides the inferred type of specific columns using SQL DDL syntax, without writing the whole schema.

(spark.readStream.format("cloudFiles")
   .option("cloudFiles.format", "csv")
   .option("header", "true")
   .option("cloudFiles.inferColumnTypes", "true")
   .option("cloudFiles.schemaHints", "order_id BIGINT, amount DECIMAL(12,2)")
   .option("cloudFiles.schemaLocation", "/Volumes/prod/bronze/_schemas/orders")
   .load("/Volumes/prod/raw/landing/orders/"))

Here order_id and amount are forced to the declared types while all the other columns are inferred. Hints are useful when the sample would infer double for a column that should be a decimal, or bigint for a code that must keep its leading zeros as a string (zip_code STRING).

You can also give a complete schema with .schema(...). Then no inference happens, and schemaLocation is not needed, but evolution is limited because you control the schema.

5. Schema enforcement and evolution

Data sources change: a producer adds a column, or renames one. Auto Loader detects such changes and reacts according to cloudFiles.schemaEvolutionMode.

Mode	New column appears	Stream behaviour	Notes
`addNewColumns` (default when no schema is given)	Schema is updated with the new column	Stream fails once with an unknown field error, then succeeds on restart with the new schema	Jobs should retry automatically
`rescue`	Schema is not evolved	Stream does not fail; new data goes to the rescued data column	Good when you want stability
`failOnNewColumns`	Not accepted	Stream fails until you update the schema or the data	Strictest option
`none` (default when you provide a schema)	Ignored unless a rescued data column is set	Stream does not fail	New columns are dropped if not rescued

Two details deserve attention:

With addNewColumns, the stream stops with an exception the first time it sees the new column, updates the schema in schemaLocation, and works after a restart. In production you run the stream inside a Lakeflow Job with retries, so the restart is automatic. This surprises beginners who expect the new column to appear without any interruption.
When you supply a schema, evolution does not happen by default; the mode is none.

Writing to a Delta table also needs mergeSchema when you want the target table to accept the evolved columns:

.writeStream.option("mergeSchema", "true")

Type changes are also handled: with Databricks Runtime 16.4 and above, the addNewColumnsWithTypeWidening mode can widen compatible types (for example int to bigint) and rescue incompatible ones. This is more advanced than the exam normally requires, but you should recognise it.

The rescued data column

The rescued data column keeps data that does not fit the schema, so nothing is silently lost. It is named _rescued_data and is added by default when Auto Loader infers the schema. It stores, as JSON, the following:

columns that are absent from the schema (new fields);
values whose type does not match the schema (for example text in a numeric column);
columns whose name differs only by case;
the source file path (_file_path).

You can rename it with cloudFiles.rescuedDataColumn or disable it by setting the option to an empty value. Typical use: keep it in bronze, then monitor it. A growing number of non-null _rescued_data rows is an alert that the producer changed the format.

SELECT _rescued_data, count(*) AS n
FROM prod.bronze.events
WHERE _rescued_data IS NOT NULL
GROUP BY _rescued_data;

Worked example: choosing an evolution mode

Scenario. A mobile app team adds new event properties almost every sprint. The data engineering team feeds a bronze table consumed by dashboards that must never break and must not stop overnight. The team still wants to review new fields before exposing them. Which mode and design do you choose?

Reasoning.

failOnNewColumns would stop the stream each time a property appears, which violates "must not stop".
addNewColumns would also stop the stream once per change and then change the bronze schema, which may break dashboards that use SELECT *.
rescue keeps the stream running, does not change the table schema, and stores unknown properties in _rescued_data as JSON.
The team reviews _rescued_data regularly and promotes useful fields with an explicit change to the pipeline.

Why other choices are weaker. none also keeps the job running but drops unknown fields if the rescued column is disabled, losing data. Removing the schema location removes the stored schema and defeats the purpose.

6. Checkpoints and triggers

Checkpoints

The checkpoint (checkpointLocation) records which files have been processed, so the stream is exactly-once at the file level and can restart without reprocessing. Rules to remember:

One checkpoint per stream; never share it between two streams or two target tables.
Do not delete it casually. Deleting the checkpoint makes Auto Loader forget what it processed, and it reloads every file, creating duplicates unless the sink deduplicates.
Changing the source path of a stream while keeping the checkpoint is not supported; start a new checkpoint.
If you change the schema location without the checkpoint, inference restarts.

Triggers

Trigger	Behaviour	When to use
default (no trigger) / `processingTime="1 minute"`	Continuously processes micro-batches	Low latency, always-on cluster
`availableNow=True`	Processes all files available now, in several micro-batches if needed, then stops	Scheduled incremental batch on jobs compute
`once=True` (legacy)	One micro-batch, then stops	Replaced by `availableNow`

availableNow is the pattern that connects streaming and batch: you keep the checkpoint and exactly-once behaviour, but you pay for compute only while the job runs. A nightly job with availableNow=True is the typical answer to "incremental, cheap, no always-on cluster".

Use .awaitTermination() after toTable(...) in a notebook when later cells must wait for the stream to finish.

7. Choosing between Auto Loader and other options

Situation	Choice
Thousands of files per day, simple SQL, no checkpoint management	`COPY INTO`
Millions of files, continuous arrival, evolving schema	Auto Loader
Hourly batch with cheap compute but incremental loading	Auto Loader with `availableNow`
Declarative pipeline with quality rules	Streaming table with `STREAM read_files` in Lakeflow Spark Declarative Pipelines
One-off file	Upload UI or `CREATE TABLE AS SELECT`

A common distractor on the exam presents COPY INTO and Auto Loader together; the discriminating words are "scalable to millions of files", "schema evolution" and "rescued data", which point to Auto Loader.

8. Operations: monitoring and tuning

Add audit columns: _metadata.file_path, _metadata.file_modification_time and current_timestamp() make each row traceable.
Limit batch size with cloudFiles.maxFilesPerTrigger or cloudFiles.maxBytesPerTrigger when a first run over a large backlog overwhelms the cluster.
Backfill historical files with cloudFiles.includeExistingFiles (default true); set it to false to process only files that arrive after the stream starts.
Allow overwrites of files only deliberately with cloudFiles.allowOverwrites; by default a modified file is not reprocessed.
Watch the stream: the streaming query progress and the Lakeflow Jobs run page show rows processed per batch and backlog.

Example. After a three-year backlog of 2 million files, the first run of a new Auto Loader stream takes 10 hours and fails with an out-of-memory error. Setting cloudFiles.maxFilesPerTrigger = 10000 splits the backlog into smaller micro-batches that fit in memory, and availableNow still stops after the backlog is drained.

Common pitfalls and exam tips

Pitfall: format("json") instead of format("cloudFiles"). The file format belongs in cloudFiles.format.
Pitfall: no schemaLocation. Inferred schemas must be stored, otherwise the stream cannot track or evolve the schema.
Pitfall: expecting typed columns from JSON or CSV. Without inferColumnTypes or hints, columns are strings.
Pitfall: deleting the checkpoint to "fix" an error. The stream reprocesses everything and creates duplicates.
Pitfall: surprise stop with addNewColumns. One failure is the intended behaviour; retries restart it with the new schema.
Pitfall: confusing _rescued_data with a corrupt record column. It also holds extra and mismatched-type fields, not only malformed rows.
Exam tip. "Many millions of files and slow listing" means file notification mode (cloudFiles.useNotifications).
Exam tip. "Incremental, scheduled, no always-on cluster" means trigger(availableNow=True).
Exam tip. "Keep running when the schema changes and do not lose new fields" means rescue mode.

Mini case study

Case. Nordic Retail's point-of-sale devices upload JSON receipts to abfss://pos@nordicdata.dfs.core.windows.net/receipts/. About 8 million files accumulate per month, in a flat folder, and the vendor adds new optional fields without warning. Finance needs the bronze table refreshed every hour, and the platform team will not pay for a cluster that runs all day. The team has permission to create storage events.

Tasks. Select the detection mode, trigger and evolution mode, and describe what to watch.

Solution.

Detection mode: file notification (cloudFiles.useNotifications = true), because a flat directory with millions of files makes listing slow, and event permissions are available.
Trigger: availableNow=True, started hourly by a Lakeflow Job on jobs compute, which gives incremental processing without an always-on cluster.
Evolution mode: addNewColumns with job retries is acceptable if finance accepts evolving columns; if bronze consumers must never change, use rescue and promote fields deliberately. Given that finance wants stable reports, choose rescue and review _rescued_data.
Schema: cloudFiles.inferColumnTypes = true plus hints for monetary columns (total DECIMAL(12,2)).
Monitoring: count non-null _rescued_data rows per day and alert when they rise; keep _metadata.file_path for traceability.

Key terms

Auto Loader: the cloudFiles streaming source for incremental file ingestion.
Checkpoint: stored progress that makes a stream restartable and exactly-once.
Schema location: folder where Auto Loader keeps the inferred schema history.
Schema hints: DDL overrides for selected column types.
Rescued data column: _rescued_data, the JSON column keeping data that does not fit the schema.
File notification mode: discovery through cloud event notifications rather than directory listing.
availableNow: trigger that processes available data and stops.

Summary checklist

I can write an Auto Loader stream with cloudFiles, cloudFiles.format, schemaLocation and a checkpoint.
I can explain directory listing versus file notification mode and when to use each.
I know JSON and CSV columns are strings unless inferColumnTypes or hints are used.
I can name the four evolution modes and describe the stream behaviour of each.
I know what _rescued_data stores.
I know why the checkpoint must not be deleted or shared.
I can run Auto Loader as scheduled incremental batch with availableNow.