Incremental file ingestion in PySpark: Structured Streaming, checkpoints and availableNow

Module 4 loaded a file you named on the command line. The shop's system does not wait for you: it drops a new file of orders into a folder whenever it has one. This module gets data in as it arrives: files first, then apps and databases, then the changes a database makes. This part loads the new files in a folder, once each, with plain PySpark.

Load only what is new

The job has 2 rules. Every line of every file reaches the table, and none arrives twice, however often the job runs. Keeping a list of loaded files yourself is easy to get wrong when a run fails halfway. Spark Structured Streaming keeps that list for you, in a folder called the checkpoint.

First make the files. Save this as make_files.py, in a new folder with delta_session.py from module 4:

"""Write the shop's order files for 2 days into landing/, as the shop's system would drop them."""
import sys
from pathlib import Path

DAYS = {
    "2026-10-01": [
        "1001,C01,placed,25.50,2026-10-01 09:15:00",
        "1002,C02,placed,12.00,2026-10-01 10:02:00",
        "1003,C03,placed,abc,2026-10-01 11:30:00",
    ],
    "2026-10-02": [
        "1001,C01,shipped,25.50,2026-10-02 08:00:00",
        "1004,C01,placed,40.00,2026-10-02 12:45:00",
    ],
}
day = sys.argv[1]
landing = Path("landing")
landing.mkdir(exist_ok=True)
lines = ["order_id,customer_id,status,amount,updated_at", *DAYS[day]]
(landing / f"orders-{day}.csv").write_text("\n".join(lines) + "\n")
print("wrote", landing / f"orders-{day}.csv")

Then the load. Save this as ingest.py:

"""Load the order files in landing/ that have not been loaded yet, then stop."""
from pyspark.sql import functions as F
from delta_session import delta_spark

spark = delta_spark("ingest")
ORDERS = "order_id BIGINT, customer_id STRING, status STRING, amount DECIMAL(10,2), updated_at TIMESTAMP, _bad STRING"

files = (
    spark.readStream.format("csv")
    .option("header", True)
    .option("columnNameOfCorruptRecord", "_bad")
    .schema(ORDERS)
    .load("landing")
    .withColumn("source_file", F.col("_metadata.file_name"))
)
query = (
    files.writeStream.format("delta")
    .option("checkpointLocation", "checkpoints/bronze_orders")
    .trigger(availableNow=True)
    .toTable("bronze_orders")
)
query.awaitTermination()

print("rows loaded:", sum(progress.numInputRows for progress in query.recentProgress))
spark.table("bronze_orders").select("order_id", "status", "amount", "source_file", "_bad") \
    .orderBy("source_file", "order_id").show(truncate=False)

Run python make_files.py 2026-10-01, then python ingest.py:

wrote landing/orders-2026-10-01.csv
rows loaded: 3
+--------+------+------+---------------------+---------------------------------------+
|order_id|status|amount|source_file          |_bad                                   |
+--------+------+------+---------------------+---------------------------------------+
|1001    |placed|25.50 |orders-2026-10-01.csv|NULL                                   |
|1002    |placed|12.00 |orders-2026-10-01.csv|NULL                                   |
|1003    |placed|NULL  |orders-2026-10-01.csv|1003,C03,placed,abc,2026-10-01 11:30:00|
+--------+------+------+---------------------+---------------------------------------+

The line with abc as its amount is kept whole in _bad, as in module 2, so nothing is lost on the way in. Now the second day: run python make_files.py 2026-10-02 and python ingest.py again:

wrote landing/orders-2026-10-02.csv
rows loaded: 2
+--------+-------+------+---------------------+---------------------------------------+
|order_id|status |amount|source_file          |_bad                                   |
+--------+-------+------+---------------------+---------------------------------------+
|1001    |placed |25.50 |orders-2026-10-01.csv|NULL                                   |
|1002    |placed |12.00 |orders-2026-10-01.csv|NULL                                   |
|1003    |placed |NULL  |orders-2026-10-01.csv|1003,C03,placed,abc,2026-10-01 11:30:00|
|1001    |shipped|25.50 |orders-2026-10-02.csv|NULL                                   |
|1004    |placed |40.00 |orders-2026-10-02.csv|NULL                                   |
+--------+-------+------+---------------------+---------------------------------------+

The folder holds 2 files, and only the new one was read. Run python ingest.py a third time and it prints rows loaded: 0. This table is bronze, the orders as they arrived, with both versions of order 1001. Picking the newest version of each order is a later step's job, as module 4 did with MERGE.

What the checkpoint holds

Look inside checkpoints/bronze_orders. sources lists every file the stream has read, with the batch that read it. offsets records what each batch was planned to read, before it ran, and commits records each batch that finished writing. A run that fails between the two is replayed from its plan on the next start, and Delta's transaction log lets the table guarantee exactly-once processing, so a replayed batch is not written twice. That is how Spark gets each line in exactly once: a source it can replay, a sink that ignores a batch it already has, and the checkpoint between them.

2 rules follow. Each stream needs its own checkpoint folder, never shared with another stream. And the checkpoint belongs with the table: delete it, or point the stream at a new folder, and the next run starts fresh and reads every file again. Part 4 adds a check that catches that.

The checkpoint knows a file by its path, and reads each file once. If the shop's system writes a file slowly, line by line, the stream can read it half-written and never come back for the rest. Spark's docs ask for files to be placed in the folder whole: write them somewhere else, then move them in.

A stream that stops

trigger(availableNow=True) makes a stream behave like a batch job: it reads everything new, in 1 or more batches, and stops. You run it on a schedule, and the checkpoint carries over from one run to the next. Without a trigger, a stream keeps running and checks the folder again and again, which costs compute all day for files that arrive a few times a day.

Name the format, too. On your computer, toTable without .format("delta") makes a Parquet table, Spark's default. On Databricks, Delta is the default for every table.

On Databricks

On serverless compute, which is all Free Edition has, availableNow is the trigger to use: Databricks supports it, and refuses the triggers that run without stopping with the error INFINITE_STREAMING_TRIGGER_NOT_SUPPORTED. Put the files and the checkpoint in volumes, such as /Volumes/<catalog>/bronze/landing and /Volumes/<catalog>/bronze/checkpoints/bronze_orders. With those paths in place of landing and checkpoints/bronze_orders, and without the import and the delta_spark line, the code above runs in a notebook. On Databricks the file source is usually replaced by Auto Loader, which part 2 covers.