Databricks Auto Loader explained: cloudFiles, schema evolution and rescued data

Part 1 loaded the shop's order files into a bronze table with plain PySpark: a streaming read of landing/, a checkpoint that remembers each file, and a trigger that stops when the new files are loaded. On Databricks, the same job is usually done by Auto Loader, a source built for files in cloud storage. This part is Databricks only, so its code follows Databricks' docs.

Auto Loader in a notebook

Auto Loader is a streaming source called cloudFiles. It records every file it finds in the checkpoint, and loads each one exactly once, as part 1's stream did. What it adds, by Databricks' account, is scale and the schema: it finds new files among billions without listing every folder, and it can infer the schema of the files and change it when they change.

In a notebook on Databricks, with the shop's schemas from module 5:

from pyspark.sql import functions as F

catalog = spark.catalog.currentCatalog()
spark.sql("CREATE VOLUME IF NOT EXISTS bronze.landing")
spark.sql("CREATE VOLUME IF NOT EXISTS bronze.checkpoints")
landing = f"/Volumes/{catalog}/bronze/landing"
checkpoint = f"/Volumes/{catalog}/bronze/checkpoints/orders_raw"

query = (
    spark.readStream.format("cloudFiles")
    .option("cloudFiles.format", "csv")
    .option("cloudFiles.schemaLocation", checkpoint)
    .option("cloudFiles.schemaHints", "order_id BIGINT, amount DECIMAL(10,2), updated_at TIMESTAMP")
    .load(landing)
    .select("*", F.col("_metadata.file_name").alias("source_file"))
    .writeStream
    .option("checkpointLocation", checkpoint)
    .option("mergeSchema", "true")
    .trigger(availableNow=True)
    .toTable("bronze.orders_raw")
)
query.awaitTermination()

Upload part 1's 2 files to the landing volume in Catalog, run the cell, and bronze.orders_raw holds their 5 rows. Run it again and nothing is added. To see which files the stream has taken, query its checkpoint:

SELECT * FROM cloud_files_state('/Volumes/<catalog>/bronze/checkpoints/orders_raw')

The schema it keeps

There is no .schema(...) in that code. Auto Loader samples the first files, up to 50 GB or 1,000 files, and writes the schema it found into cloudFiles.schemaLocation, here the same folder as the checkpoint, as Databricks' examples do. For CSV, JSON and XML, which do not record types, it infers every column as a string, so a later file with an odd value never breaks the stream. cloudFiles.inferColumnTypes set to true infers types from the sample instead, and cloudFiles.schemaHints, used above, names the types you already know.

A value that does not fit goes into a column Auto Loader adds itself, _rescued_data, as JSON with the path of its file. Order 1003's amount abc cannot be a DECIMAL(10,2), so its amount is null and the abc is kept in _rescued_data. That is part 1's _bad column, kept for you.

When a file brings a new column

Say the shop's system starts sending a coupon column. What Auto Loader does is set by cloudFiles.schemaEvolutionMode, and each mode is a different answer to the question "who decides about a new column?":

{"order_id": 1005, "status": "placed", "amount": 18.25, "coupon": "SAVE10"}

The default when Auto Loader infers the schema is addNewColumns. It stops the stream once, on purpose, after recording the new column, and the next start loads the file with it. Databricks recommends running the stream as a job that restarts on its own, and the mergeSchema option on the write lets the Delta table take the new column too.

Give Auto Loader a full schema with .schema(...) instead, and the default becomes none, which ignores new columns. That is the trap in the table: a new column is silently dropped. For a schema you know, Databricks' pattern is rescue mode, which keeps anything unexpected in _rescued_data.

Finding the files

Auto Loader finds new files in 1 of 2 ways. Directory listing, the default, lists the folder on each run. File notification has the cloud storage announce each new file, and Databricks recommends it, with file events, for most workloads. Auto Loader does not promise to load files in the order they arrived, which matters for changes to the same order, as part 3 shows.

Apps and databases: Lakeflow Connect

Not every source drops files. The shop's payments might live in Salesforce, its stock in PostgreSQL. Lakeflow Connect has managed connectors that load these into Unity Catalog tables with no files in between. There are connectors for SaaS apps such as Salesforce, Workday, HubSpot and Jira, and for databases such as MySQL, PostgreSQL and SQL Server. There are also file stores such as SharePoint and Google Drive, and message buses.

Each connector builds an ingestion pipeline on serverless compute, governed by Unity Catalog. Its first run loads all the data you selected, and later runs load only what changed, where the source can say what changed. The database connectors read the database's own record of changes, through an ingestion gateway, which is the subject of part 3. Free Edition limits outbound internet access to a set of trusted domains, so a connector to an outside system may not reach it there; files in a volume always work.