What happens if I delete the checkpoint?

Auto Loader forgets which files it processed and will load them all again, creating duplicates in an append table. Keep checkpoints with the pipeline, and do not delete them unless you want a full reload into a cleaned table.

Databricks Module 5: Ingesting Files with Auto Loader and COPY INTO

Introduction

Most pipelines start by ingesting files that keep arriving in cloud storage. Auto Loader detects new files and processes each exactly once without you tracking them. This module sets it up into a bronze table, and shows COPY INTO for simpler cases.

Prerequisites

Module 4 completed
A volume path where you can put files (for example /Volumes/lab/sales/raw_files/)

Step-by-Step Guide

Step 1: Land sample files

Create a folder in your volume and add a few small JSON files, as if a source system delivered them. In a notebook, you can write them with dbutils.

import json

base = "/Volumes/lab/sales/raw_files/orders/"
dbutils.fs.mkdirs(base)

batch1 = [{"order_id": 1, "customer_id": 101, "amount": 120.5, "status": "SHIPPED"},
          {"order_id": 2, "customer_id": 102, "amount": 80.0, "status": "PENDING"}]
dbutils.fs.put(base + "orders_001.json", "\n".join(json.dumps(r) for r in batch1), True)

Step 2: Read the files incrementally with Auto Loader

Auto Loader is the cloudFiles source for Structured Streaming. It tracks which files it has already processed in a checkpoint, infers the schema, and scales to millions of files. The schemaLocation option stores the inferred schema so it can evolve.

raw = (spark.readStream.format("cloudFiles")
       .option("cloudFiles.format", "json")
       .option("cloudFiles.schemaLocation", "/Volumes/lab/sales/raw_files/_schemas/orders")
       .load("/Volumes/lab/sales/raw_files/orders/"))

Step 3: Write to a bronze Delta table

Write the stream to a Delta table with a checkpoint location. The availableNow trigger processes everything currently available and then stops, which gives batch-style scheduled runs with streaming's file tracking. Add metadata columns such as the source file and the load time.

from pyspark.sql import functions as F

(raw.withColumn("_source_file", F.col("_metadata.file_path"))
    .withColumn("_ingested_at", F.current_timestamp())
    .writeStream
    .option("checkpointLocation", "/Volumes/lab/sales/raw_files/_checkpoints/orders_bronze")
    .trigger(availableNow=True)
    .toTable("lab.sales.orders_bronze"))

Step 4: Add more files and run again

Drop a second file with a new column, then run the same cells again. Auto Loader picks up only the new file. By default, when it meets a new column it can stop the stream and record the new schema, and the next run succeeds, which protects you from silent surprises. You can configure schema evolution modes and a rescued data column to keep unexpected fields.

batch2 = [{"order_id": 3, "customer_id": 101, "amount": 45.25, "status": "SHIPPED", "coupon": "WELCOME"}]
dbutils.fs.put(base + "orders_002.json", json.dumps(batch2[0]), True)

# run the readStream/writeStream cells again, then:
display(spark.table("lab.sales.orders_bronze"))

Step 5: Use COPY INTO for simple batch loads

COPY INTO loads files from a location into a Delta table and skips files it has already loaded, which makes it idempotent. It is simpler than Auto Loader for small, occasional loads, while Auto Loader is better for large or continuous file arrival. Check the documentation for which is recommended for your case.

COPY INTO lab.sales.customers_raw
FROM '/Volumes/lab/sales/raw_files/customers/'
FILEFORMAT = CSV
FORMAT_OPTIONS ('header' = 'true', 'inferSchema' = 'true')
COPY_OPTIONS ('mergeSchema' = 'true');

FAQs

What happens if I delete the checkpoint?: Auto Loader forgets which files it processed and will load them all again, creating duplicates in an append table. Keep checkpoints with the pipeline, and do not delete them unless you want a full reload into a cleaned table.

Summary

Auto Loader reads only new files using the cloudFiles source, a schema location, and a checkpoint, writes to a bronze Delta table with metadata columns, handles schema changes safely, and COPY INTO covers simpler batch loads.