Incremental ingestion without the plumbing: an Auto Loader (cloudFiles) pipeline built end to end

Auto Loader (cloudFiles) incrementally ingests new files from cloud storage exactly once. It tracks what it has already processed in a checkpoint. You never write "which files are new?" logic yourself, and reruns are safe.

#The scenario

An order system drops JSON files into abfss://landing@stlakeprod.dfs.core.windows.net/orders/ every few minutes. We want:

• bronze.orders_raw: every record as it arrived, plus file metadata.
• silver.orders: typed, de-duplicated and validated.
• gold.daily_sales: aggregated for BI.

#Step 0: prerequisites

-- external location over the landing container (storage credential created by an admin)
CREATE EXTERNAL LOCATION IF NOT EXISTS landing_orders
  URL 'abfss://landing@stlakeprod.dfs.core.windows.net/orders'
  WITH (STORAGE CREDENTIAL lake_mi);
GRANT READ FILES ON EXTERNAL LOCATION landing_orders TO `data-engineers`;

-- a volume for checkpoints and schema tracking
CREATE VOLUME IF NOT EXISTS ops.pipelines.autoloader;

#Step 1: bronze with Auto Loader

src  = "abfss://landing@stlakeprod.dfs.core.windows.net/orders/"
base = "/Volumes/ops/pipelines/autoloader/orders"

bronze = (spark.readStream.format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", f"{base}/_schema")      # inferred schema is stored and versioned here
    .option("cloudFiles.inferColumnTypes", "true")               # otherwise JSON columns arrive as strings
    .option("cloudFiles.schemaHints", "order_id STRING, amount DECIMAL(18,2), order_ts TIMESTAMP")
    .option("cloudFiles.schemaEvolutionMode", "addNewColumns")   # default
    .option("cloudFiles.maxFilesPerTrigger", 1000)
    .load(src)
    .selectExpr("*", "_metadata.file_path AS source_file", "_metadata.file_modification_time AS file_ts",
                "current_timestamp() AS ingest_ts"))

(bronze.writeStream
    .option("checkpointLocation", f"{base}/_checkpoint_bronze")
    .option("mergeSchema", "true")
    .trigger(availableNow=True)          # process everything new, then stop: perfect for scheduled jobs
    .toTable("main.bronze.orders_raw"))

What happens:

• The first run lists the folder, infers the schema from a sample and writes it to _schema.
• Records that don't fit the schema go to _rescued_data instead of failing the stream.
• When a new column appears, the stream stops with an UnknownFieldException and records the new schema. The next run picks it up. Set your job to retry once so this is self-healing.

#Schema evolution modes

#Step 2: silver, typed and de-duplicated

from pyspark.sql import functions as F

silver = (spark.readStream.table("main.bronze.orders_raw")
    .where("order_id IS NOT NULL AND _rescued_data IS NULL")
    .withColumn("order_date", F.to_date("order_ts"))
    .withWatermark("order_ts", "2 hours")
    .dropDuplicatesWithinWatermark(["order_id"]))

(silver.writeStream
    .option("checkpointLocation", f"{base}/_checkpoint_silver")
    .trigger(availableNow=True)
    .toTable("main.silver.orders"))

Send rows with _rescued_data to a quarantine table and alert on them. Don't drop them silently.

If orders can be updated, use foreachBatch with a MERGE instead of append:

def upsert(batch_df, batch_id):
    batch_df.createOrReplaceTempView("updates")
    batch_df.sparkSession.sql("""
      MERGE INTO main.silver.orders t
      USING (SELECT * FROM updates QUALIFY row_number() OVER (PARTITION BY order_id ORDER BY order_ts DESC) = 1) s
      ON t.order_id = s.order_id
      WHEN MATCHED AND s.order_ts > t.order_ts THEN UPDATE SET *
      WHEN NOT MATCHED THEN INSERT *""")

silver.writeStream.foreachBatch(upsert).option("checkpointLocation", f"{base}/_checkpoint_silver_merge") \
    .trigger(availableNow=True).start()

#Step 3: gold

CREATE OR REPLACE TABLE main.gold.daily_sales AS
SELECT order_date, region, count(*) AS orders, sum(amount) AS revenue
FROM main.silver.orders GROUP BY order_date, region;

For large data, use a materialized view in a Lakeflow pipeline or a SQL warehouse instead. It refreshes incrementally where possible.

#Step 4: schedule it

Create a Lakeflow job with three tasks (bronze, then silver, then gold) on a 15-minute schedule or a file arrival trigger on the landing location. Give it 1–2 retries, so a schema-change stop restarts automatically, and alert on failure.

#The same pipeline as a Lakeflow declarative pipeline (SQL)

CREATE OR REFRESH STREAMING TABLE orders_raw AS
SELECT *, _metadata.file_path AS source_file, current_timestamp() AS ingest_ts
FROM STREAM read_files('abfss://landing@stlakeprod.dfs.core.windows.net/orders/',
                       format => 'json', schemaHints => 'amount DECIMAL(18,2)');

CREATE OR REFRESH STREAMING TABLE orders (
  CONSTRAINT valid_id EXPECT (order_id IS NOT NULL) ON VIOLATION DROP ROW
) AS SELECT * FROM STREAM(orders_raw) WHERE _rescued_data IS NULL;

CREATE OR REFRESH MATERIALIZED VIEW daily_sales AS
SELECT to_date(order_ts) AS order_date, region, count(*) orders, sum(amount) revenue
FROM orders GROUP BY ALL;

The pipeline manages checkpoints, schema locations and retries for you. It's the simplest option for new work.

#Directory listing vs file notification

Newer runtimes support managed file events (cloudFiles.useManagedFileEvents) on Unity Catalog external locations. They replace the per-stream queue setup of cloudFiles.useNotifications. Check that your workspace has file events enabled first.

#Monitoring

-- which files has the stream processed?
SELECT * FROM cloud_files_state('/Volumes/ops/pipelines/autoloader/orders/_checkpoint_bronze') ORDER BY commit_time DESC LIMIT 20;

• In a notebook, query.lastProgress shows numInputRows and numFilesOutstanding (backlog) in the source metrics.
• Alert on backlog growth, on _rescued_data counts and on job failures.

#Production problems and fixes

Warning

A checkpoint belongs to exactly one stream writing to one table. Copying, sharing or deleting it causes re-ingestion or data loss. Keep checkpoints in a dedicated volume and protect them like table data.