Ingesting Files with Auto Loader and COPY INTO

Both tools solve the same problem — load files that keep arriving, exactly once — and both remember what they have already read. The difference is scale and how much they do about a schema that changes.

COPY INTO

create table if not exists bookshop.raw.orders_landed (
    order_id    bigint,
    customer_id bigint,
    ordered_at  date,
    status      string,
    amount      decimal(10,2)
);

copy into bookshop.raw.orders_landed
from '/Volumes/bookshop/raw/landing/'
fileformat = csv
format_options ('header' = 'true', 'inferSchema' = 'false')
copy_options ('mergeSchema' = 'false');

num_affected_rows  num_inserted_rows  num_skipped_corrupt_files
-----------------  -----------------  -------------------------
             9812               9812                          0

Run it again with no new files:

num_affected_rows  num_inserted_rows  num_skipped_corrupt_files
-----------------  -----------------  -------------------------
                0                  0                          0

Zero. COPY INTO records the files it has loaded against the target table, so re-running is safe — the same idempotence dbt’s incremental models and Snowflake’s COPY INTO give you, and the reason a retried job does not duplicate rows.

Add a file to the volume and run once more:

num_affected_rows  num_inserted_rows  num_skipped_corrupt_files
-----------------  -----------------  -------------------------
             4120               4120                          0

Only the new file. That makes COPY INTO a complete incremental loader in one statement, which is often all a daily batch needs.

Auto Loader

For continuous arrival, or millions of files, the streaming source scales better:

from pyspark.sql.functions import current_timestamp, input_file_name

(spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "csv")
    .option("cloudFiles.schemaLocation", "/Volumes/bookshop/raw/_schemas/orders")
    .option("header", "true")
    .load("/Volumes/bookshop/raw/landing/")
    .withColumn("_ingested_at", current_timestamp())
    .withColumn("_source_file", input_file_name())
 .writeStream
    .option("checkpointLocation", "/Volumes/bookshop/raw/_checkpoints/orders")
    .trigger(availableNow=True)
    .toTable("bookshop.raw.orders_autoloaded"))

{
  "id": "8f2c1a44-8e21-4b0e-9a3c-1d84f0b27a51",
  "runId": "3d9c4b21-77a1-4e02-b8f1-5c2e91a4d883",
  "batchId": 0,
  "numInputRows": 13932,
  "inputRowsPerSecond": 0.0,
  "processedRowsPerSecond": 4204.11,
  "sources": [{
    "description": "CloudFilesSource[/Volumes/bookshop/raw/landing/]",
    "numFilesProcessed": 2,
    "numBytesProcessed": 242704
  }]
}

Two options carry the weight here:

• schemaLocation — where the inferred schema is stored between runs, so column order and types stay stable and evolution can be detected.
• checkpointLocation — what has been processed. Treat it as production state: deleting it reprocesses every file in the directory, which on an append target means duplicates.

trigger(availableNow=True) processes everything waiting and stops, which is how you get incremental semantics on a schedule without paying for an always-on stream. Drop it for a continuous stream, or use processingTime for micro-batches.

spark.table("bookshop.raw.orders_autoloaded").select("order_id", "amount", "_source_file").show(3, truncate=60)

+--------+------+------------------------------------------------------------+
|order_id|amount|                                                _source_file|
+--------+------+------------------------------------------------------------+
|    1001| 25.50|/Volumes/bookshop/raw/landing/orders_2026_01.csv             |
|    1002| 12.00|/Volumes/bookshop/raw/landing/orders_2026_01.csv             |
|    1003| 40.00|/Volumes/bookshop/raw/landing/orders_2026_01.csv             |
+--------+------+------------------------------------------------------------+
only showing top 3 rows

Recording the source file costs one column and answers “where did this row come from” for the life of the table.

Rescued data

Point Auto Loader at a file where amount is sometimes text:

(spark.readStream.format("cloudFiles")
    .option("cloudFiles.format", "csv")
    .option("cloudFiles.schemaLocation", "/Volumes/bookshop/raw/_schemas/orders")
    .option("cloudFiles.schemaHints", "order_id bigint, amount decimal(10,2)")
    .option("header", "true")
    .load("/Volumes/bookshop/raw/landing/")
 .writeStream
    .option("checkpointLocation", "/Volumes/bookshop/raw/_checkpoints/orders")
    .trigger(availableNow=True)
    .toTable("bookshop.raw.orders_autoloaded"))

select order_id, amount, _rescued_data
from bookshop.raw.orders_autoloaded
where _rescued_data is not null
limit 3;

order_id  amount  _rescued_data
--------  ------  --------------------------------------------------------------
    1412    NULL  {"amount":"n/a","_file_path":"/Volumes/.../orders_2026_01.csv"}
    3980    NULL  {"amount":"n/a","_file_path":"/Volumes/.../orders_2026_01.csv"}
    4771    NULL  {"amount":"","_file_path":"/Volumes/.../orders_2026_01.csv"}

The rows loaded, the bad values were preserved as JSON, and nothing failed. This is the behaviour to want in a bronze layer: never lose a record, never let a bad value masquerade as a good one. Alert on _rescued_data is not null and the malformed rows become a ticket rather than a silent gap.

schemaHints pins the types you care about and lets Auto Loader infer the rest — better than full inference, which can decide a column is a string because of one bad file, and better than a full schema, which you then have to maintain.

Schema evolution

An upstream team adds a channel column mid-month:

StreamingQueryException: Encountered unknown field(s) during parsing: {"channel":"web"}

A schema mismatch detected when writing to the Delta table.
The stream has been stopped so the new schema can be picked up.
Restart the stream to continue.

The stream stops on purpose. cloudFiles.schemaEvolutionMode decides what happens:

The default surprises people, but it is the right trade: a job orchestrated with retries restarts, picks up the new column, and continues, while the failure is visible in the run history. rescue is the choice when the stream must never stop and you will reconcile later.

    .option("cloudFiles.schemaEvolutionMode", "rescue")

select count(*) as rescued from bookshop.raw.orders_autoloaded where _rescued_data is not null;

rescued
-------
   4120

Finding files at scale

By default Auto Loader lists the directory. Past a few hundred thousand files that gets slow, so switch to file notifications — the cloud provider tells Databricks about each new object:

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

Directory listing is simpler and needs no extra cloud permissions; notifications need rights to create a queue and subscription but scale to millions of files. Start with listing and switch when listing time shows up in your batch duration.

Two more options worth knowing:

    .option("cloudFiles.maxFilesPerTrigger", "1000")     # bound each batch
    .option("cloudFiles.includeExistingFiles", "false")  # only files arriving from now on

Watching a stream

for q in spark.streams.active:
    print(q.name, q.status["message"], q.lastProgress["numInputRows"])

orders_ingest Waiting for data to arrive 0

describe history bookshop.raw.orders_autoloaded limit 3;

version  operation           operationMetrics
-------  ------------------  --------------------------------------------
     12  STREAMING UPDATE    {numOutputRows: 4120, numAddedFiles: 3}
     11  STREAMING UPDATE    {numOutputRows: 9812, numAddedFiles: 7}
     10  CREATE TABLE        {}

Each micro-batch is a Delta commit, so the ingestion history is the table history — including time travel back to before a bad batch.

Practice

num_inserted_rows
-----------------
             9812

num_inserted_rows
-----------------
                0

"numFilesProcessed": 9,
"numInputRows": 13932

order_id  amount  _rescued_data
--------  ------  --------------------------------
    1412    NULL  {"amount":"n/a","_file_path":...}

-- first run
StreamingQueryException: Encountered unknown field(s) during parsing: {"channel":"web"}
-- after restart
"numInputRows": 4120

Next: the medallion architecture — bronze, silver and gold, and what belongs in each.