Databricks Auto Loader: Incremental File Loading Without the Bookkeeping

When I built file ingestion on Databricks a few years ago, Azure Data Factory did the "which files are new?" work: an event trigger or a Get Metadata activity found the file, and a notebook processed it. It worked, but the state of what had been loaded lived in two places, and replaying a day's files was always fiddly.

Auto Loader moves that job into Databricks. It watches a cloud storage folder, keeps track of which files it has already processed, and loads only the new ones, exactly once. This post walks through it using a common insurance case: a regular feed of claims transactions arriving as CSV files.

The basic pattern

Auto Loader is a Structured Streaming source called cloudFiles. Don't let "streaming" put you off: with availableNow it runs like a batch job, processes everything new, and stops.

 1: source_path     = "abfss://feeds@mystorage.dfs.core.windows.net/claims/"
 2: schema_path     = "abfss://meta@mystorage.dfs.core.windows.net/autoloader/claims/schema"
 3: checkpoint_path = "abfss://meta@mystorage.dfs.core.windows.net/autoloader/claims/checkpoint"
 4: 
 5: claims = (spark.readStream
 6:             .format("cloudFiles")
 7:             .option("cloudFiles.format", "csv")
 8:             .option("cloudFiles.schemaLocation", schema_path)
 9:             .option("header", True)
10:             .load(source_path))
11: 
12: (claims.writeStream
13:        .option("checkpointLocation", checkpoint_path)
14:        .trigger(availableNow=True)
15:        .toTable("prod.raw.claims_raw"))

Two locations matter:

• Checkpoint: where Auto Loader records which files it has processed. Delete it and everything loads again.
• Schema location: where it stores the schema it has inferred, and how that schema changes over time.

Schema: strings by default

For CSV and JSON, Auto Loader infers every column as STRING unless you set cloudFiles.inferColumnTypes to true. For external feeds, that default is exactly right. Land the data as text, then cast it downstream where you can handle bad values deliberately. If a few columns are safe to type up front, use schema hints:

1: .option("cloudFiles.schemaHints", "claim_ref STRING, transaction_date DATE")

When the feed adds a column

Feeds change. One month the claims file gains a cause_of_loss column. Auto Loader's default schema evolution mode, addNewColumns, handles this in two steps: the stream stops with an error that reports the new column, updates the stored schema, and the next run picks the column up. In a scheduled job with retries, that's usually invisible.

Data that doesn't fit the schema isn't thrown away. It lands in a _rescued_data column as JSON, together with the source file path:

1: SELECT _rescued_data, COUNT(*)
2: FROM   prod.raw.claims_raw
3: WHERE  _rescued_data IS NOT NULL
4: GROUP  BY _rescued_data;

If you'd rather the schema never changes on its own, set cloudFiles.schemaEvolutionMode to rescue. New columns then go to _rescued_data and the table stays as it is until you decide otherwise.

Add file metadata

For audit and replay, I always keep the source file and load time on every row:

 1: from pyspark.sql import functions as F
 2: 
 3: claims = (spark.readStream
 4:             .format("cloudFiles")
 5:             .option("cloudFiles.format", "csv")
 6:             .option("cloudFiles.schemaLocation", schema_path)
 7:             .option("header", True)
 8:             .load(source_path)
 9:             .withColumn("source_file", F.col("_metadata.file_path"))
10:             .withColumn("file_modified", F.col("_metadata.file_modification_time"))
11:             .withColumn("loaded_at", F.current_timestamp()))

The _metadata column is available on newer runtimes and is the replacement for input_file_name().

Directory listing or file notifications?

By default, Auto Loader lists the folder to find new files. That's fine for most feeds, especially when files are organised into date folders. For folders with very large numbers of files, file notification mode (cloudFiles.useNotifications) subscribes to storage events through a queue instead of listing. It needs extra permissions to create the queue and subscription, so I'd start with listing and only switch if listing becomes slow.

Upserting into a transactions table

The raw table is append-only, which is what you want for an audit trail. To maintain a current claims position, use foreachBatch to MERGE each micro-batch into a Delta table:

 1: from delta.tables import DeltaTable
 2: from pyspark.sql import Window
 3: 
 4: def upsert_claims(batch_df, batch_id):
 5:     w = Window.partitionBy("claim_ref", "bordereau_month").orderBy(F.col("file_modified").desc())
 6:     latest = (batch_df.withColumn("rn", F.row_number().over(w))
 7:                       .filter("rn = 1").drop("rn"))
 8: 
 9:     target = DeltaTable.forName(batch_df.sparkSession, "prod.core.claims")
10:     (target.alias("t")
11:        .merge(latest.alias("s"),
12:               "t.claim_ref = s.claim_ref AND t.bordereau_month = s.bordereau_month")
13:        .whenMatchedUpdateAll()
14:        .whenNotMatchedInsertAll()
15:        .execute())
16: 
17: (claims.writeStream
18:        .foreachBatch(upsert_claims)
19:        .option("checkpointLocation", checkpoint_path + "_core")
20:        .trigger(availableNow=True)
21:        .start())

The deduplication inside the batch matters. If the same claim arrives twice in one run, MERGE fails with "multiple source rows matched". Keeping the latest version per key avoids that.

Replaying and backfilling

• Reprocess everything: write to a new checkpoint location (and usually a fresh target table).
• Load only from a date: cloudFiles.includeExistingFiles set to false skips files already in the folder when the stream first starts. Use it when onboarding a feed with years of history you'll load separately.
• A corrected file with the same name: by default, a file with the same path isn't processed twice. Agree versioned file names with the feed provider, or look at cloudFiles.allowOverwrites with care.

Why I'd use it now

Compared with the ADF-triggers-a-notebook approach, Auto Loader keeps file tracking, schema changes and loading in one place, gives exactly-once processing, and makes a replay as simple as a new checkpoint. Orchestration tools still have a job, scheduling and alerting, but working out which files are new is no longer one of them.