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

To load only new files, read the folder with spark.readStream, write with a checkpoint location, and set trigger(availableNow=True). The checkpoint records every file the stream has read, so each run loads only the files added since the last one and then stops. Run it after every new file, or on a schedule, and every line reaches the table once. Delete the checkpoint and the stream starts again from nothing.

- Page: https://tecyfy.com/data-engineering/course/getting-data-in
- Course: Data engineering course (https://tecyfy.com/data-engineering/course), Module 6: Getting data in, Load new files once, with a checkpoint
- Parts of this module: Load new files once, with a checkpoint (https://tecyfy.com/data-engineering/course/getting-data-in); Auto Loader and Lakeflow Connect (https://tecyfy.com/data-engineering/course/getting-data-in/auto-loader); Change data capture with MERGE (https://tecyfy.com/data-engineering/course/getting-data-in/change-data-capture); Build it: ingest every file once (https://tecyfy.com/data-engineering/course/getting-data-in/build-it)

At the end of this part you can:

- Load only the files that are new since the last run, with a streaming read and a checkpoint
- Say what the checkpoint holds, and what happens when it is lost or shared
- Run a stream as a batch job that stops on its own, the way serverless compute requires

[Module 4](/data-engineering/course/delta-lake-and-iceberg/build-it) 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](/data-engineering/course/delta-lake-and-iceberg):

```python
"""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`:

```python
"""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`:

```text
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](/data-engineering/course/pyspark-dataframes), 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:

```text
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](/data-engineering/course/delta-lake-and-iceberg/merge) 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](/data-engineering/course/getting-data-in/build-it) 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](/data-engineering/course/getting-data-in/auto-loader) covers.

## Check yourself

1. ingest.py loaded 2 days of files. You run it again, and no file has arrived since. What happens?

   A. It loads both files again, because a stream rereads its folder on every run
   B. It fails, because availableNow needs at least 1 new file
   C. It reads nothing, writes nothing and stops, because the checkpoint lists both files as read
   D. It waits until a new file arrives

2. The shop's system writes each order file slowly, line by line, straight into landing/. What risk does that bring to ingest.py?

   A. None, because the checkpoint waits for each file to finish
   B. The stream can pick up a half-written file and never read the rest, since a file is read once
   C. The stream reads each line twice
   D. Spark refuses to read a file that is still open

### Answers

1. C. The checkpoint holds the list of files the stream has read and the batches it committed. With no new file there is nothing to read, so the run makes no batch and stops. A stream rereads files only when it starts with an empty or new checkpoint.
2. B. Spark's docs say files must be placed in the folder atomically, which on most file systems means writing them elsewhere and then moving them in. A file is read once, by name, so lines added after it was read are never loaded.

## Questions

### How do I load only new files with PySpark?

Read the folder with spark.readStream and a schema, write it with writeStream, a checkpointLocation and trigger(availableNow=True), and run it after new files arrive. The checkpoint keeps the list of files already read, so each run loads only the new ones and then stops. Keep 1 checkpoint folder for each stream, and never delete it while the table it fills still matters.

### What is a checkpoint in Spark Structured Streaming?

A folder where a streaming query records its progress, the input it read in each batch and the batches its output committed. When the query starts again with the same checkpoint, it continues where it stopped, which is what makes each file load once. A new or deleted checkpoint makes the query start fresh.

## Sources

- [Structured Streaming programming guide: APIs on DataFrames and Datasets](https://spark.apache.org/docs/latest/streaming/apis-on-dataframes-and-datasets.html)
- [Structured Streaming programming guide: getting started](https://spark.apache.org/docs/latest/streaming/getting-started.html)
- [Structured Streaming checkpoints](https://docs.databricks.com/aws/en/structured-streaming/checkpoints)
- [Serverless compute limitations](https://docs.databricks.com/aws/en/compute/serverless/limitations)
- [What is Delta Lake in Databricks?](https://docs.databricks.com/aws/en/delta/)
- [Table streaming reads and writes](https://docs.delta.io/latest/delta-streaming.html)
- [Databricks Free Edition limitations](https://docs.databricks.com/aws/en/getting-started/free-edition-limitations)
- [File metadata column](https://docs.databricks.com/aws/en/ingestion/file-metadata-column)