# Build an incremental ingestion job in PySpark that loads every file once

An ingestion job is trustworthy when it loads only new files, refuses files it cannot read correctly, and proves after each run that nothing was lost or doubled. ingest_orders.py does all 3 for the shop: a streaming read with a checkpoint and availableNow loads the new files, enforceSchema set to false stops the run on a file whose header does not match, and a check compares the lines of every file with the rows the table holds for it.

- Page: https://tecyfy.com/data-engineering/course/getting-data-in/build-it
- Course: Data engineering course (https://tecyfy.com/data-engineering/course), Module 6: Getting data in, Build it: ingest every file once
- 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:

- Write an ingestion job that loads each new file once and can run on a schedule
- Stop a load on a file with the wrong header, and recover without losing the file
- Check after every run that every line reached the table exactly once

The module so far loaded new files with a stream and a checkpoint, met Auto Loader and its schema modes, and applied a database's changes with MERGE. This part is the ingestion step of the shop's pipeline as 1 script you can run on a schedule. It needs `delta_session.py` from [module 4](/data-engineering/course/delta-lake-and-iceberg) and `make_files.py` from [part 1](/data-engineering/course/getting-data-in).

## Build it: load every file once, and prove it

Part 1's `ingest.py` loaded new files, but 2 failures would pass without a sound. A file with its columns in another order loads with the values in the wrong columns, because Spark applies a schema you give it and ignores the header. And nothing checks that the table holds what the files hold. Save this as `ingest_orders.py`, in a new folder with the 2 files named above:

```python
"""Load every new order file in landing/ into the bronze table, once, and check nothing was lost."""
from pyspark.sql import functions as F
from delta_session import delta_spark

LANDING = "landing"
CHECKPOINT = "checkpoints/bronze_orders"
TABLE = "bronze_orders"
ORDERS = "order_id BIGINT, customer_id STRING, status STRING, amount DECIMAL(10,2), updated_at TIMESTAMP"

spark = delta_spark("ingest")


def ingest():
    """Read only the files the checkpoint has not seen, append them, then stop."""
    files = (
        spark.readStream.format("csv")
        .option("header", True)
        .option("enforceSchema", False)        # fail on a header that does not match ORDERS
        .option("columnNameOfCorruptRecord", "_bad")
        .schema(f"{ORDERS}, _bad STRING")      # a line that does not parse keeps its text in _bad
        .load(LANDING)
        .select("*", F.col("_metadata.file_name").alias("source_file"), F.current_timestamp().alias("loaded_at"))
    )
    query = (
        files.writeStream.format("delta")
        .option("checkpointLocation", CHECKPOINT)
        .trigger(availableNow=True)
        .toTable(TABLE)
    )
    query.awaitTermination()
    return sum(progress.numInputRows for progress in query.recentProgress)


def check():
    """Every line of every file in landing/ is in the table exactly once."""
    lines = (
        spark.read.text(LANDING)
        .select(F.col("_metadata.file_name").alias("source_file"), "value")
        .filter(~F.col("value").startswith("order_id,"))
        .groupBy("source_file").agg(F.count("*").alias("lines"))
    )
    loaded = spark.table(TABLE).groupBy("source_file").agg(F.count("*").alias("rows"))
    report = lines.join(loaded, "source_file", "full").orderBy("source_file")
    report.show(truncate=False)
    wrong = report.filter("lines IS NULL OR rows IS NULL OR lines != rows").count()
    if wrong:
        raise RuntimeError(f"{wrong} files are missing from {TABLE} or loaded more than once")


print("new rows:", ingest())
check()
```

Run `python make_files.py 2026-10-01` and `python ingest_orders.py`, then the same for `2026-10-02`. The second run prints:

```text
new rows: 2
+---------------------+-----+----+
|source_file          |lines|rows|
+---------------------+-----+----+
|orders-2026-10-01.csv|3    |3   |
|orders-2026-10-02.csv|2    |2   |
+---------------------+-----+----+
```

Run it a third time and it prints `new rows: 0` and the same report. The check reads every file again as plain text, so it does not trust the stream: it counts each file's lines without the header, counts the table's rows for that file, and fails if a file is missing from either side or the counts differ. The line of order 1003 with `abc` counts as loaded, because it is in the table, in `_bad`. `loaded_at` records when each row arrived, which the next layer of the pipeline uses to find what is new.

## When a file is bad

Make a file with 2 columns swapped, as a changed export might:

```bash
printf 'order_id,status,customer_id,amount,updated_at\n1005,placed,C02,18.25,2026-10-03 16:20:00\n' > landing/orders-2026-10-03.csv
```

Run `python ingest_orders.py`. It stops with a `StreamingQueryException`, and in its details is the reason:

```text
CSV header does not conform to the schema.
 Header: order_id, status, customer_id, amount, updated_at
 Schema: order_id, customer_id, status, amount, updated_at
Expected: customer_id but found: status
```

That is `enforceSchema` set to `false`. With Spark's default, `true`, the file loads and order 1005 gets `placed` as its customer and `C02` as its status, a [schema drift](https://tecyfy.com/concepts/schema-drift) nobody hears about.

Now recover. Fix the file **where it is**, with the header in the right order, and run again: it prints `new rows: 1` and 3 files in the report. The failed run had planned a batch with this file, and the restart replays that batch, reading the file as it is now. Do not delete the bad file to get past it: the replayed batch then finishes without it, the checkpoint counts that file name as read, and a corrected copy under the same name is never loaded. The check catches that too, with `rows` empty for the file, but fixing in place avoids it.

## When the checkpoint is lost

Change `CHECKPOINT` to `"checkpoints/bronze_orders_v2"`, as if the folder had been deleted, and run again:

```text
new rows: 6
+---------------------+-----+----+
|source_file          |lines|rows|
+---------------------+-----+----+
|orders-2026-10-01.csv|3    |6   |
|orders-2026-10-02.csv|2    |4   |
|orders-2026-10-03.csv|1    |2   |
+---------------------+-----+----+

RuntimeError: 3 files are missing from bronze_orders or loaded more than once
```

A new checkpoint knows no files, so the stream loaded all 3 again, and every count is double. Without the check, the next run would succeed quietly on a table with every order twice. The fix is to rebuild: run `DROP TABLE bronze_orders`, delete the new checkpoint folder, keep the files, and run once more. It prints `new rows: 6` and every file once, because table and checkpoint now start together.

## On Databricks

In a notebook, leave out the import and the `delta_spark` line, and set `LANDING` and `CHECKPOINT` to volume paths, such as `/Volumes/<catalog>/bronze/landing` and `/Volumes/<catalog>/bronze/checkpoints/bronze_orders`. The check runs as it is. With Auto Loader from [part 2](/data-engineering/course/getting-data-in/auto-loader) inferring the schema instead, the swapped file is not an error at all: Auto Loader reads each CSV file by its own header, so `status` and `customer_id` land in the right columns. Run the notebook as a job on a schedule, and every run loads the new files and proves the table matches them.

Next, module 7 takes this bronze table and builds silver and gold on it with Spark Declarative Pipelines.

- Tool: [Cron expression builder](https://tecyfy.com/data-engineering/cron-builder). A schedule in words in; the cron, Quartz and Snowflake expressions, the code and the next runs out.

## Check yourself

1. ingest_orders.py stopped on orders-2026-10-03.csv, whose header has 2 columns swapped. To recover, you delete the file, run the job, then put a corrected copy back under the same name. What happens?

   A. The corrected file is loaded, because it is a new file
   B. The corrected file is never loaded, because the run after the delete marked that name as read; the check reports it as missing
   C. The job fails again on the old header
   D. Both versions of the file are loaded

2. The check reports lines 3 and rows 6 for orders-2026-10-01.csv. What is the most likely cause?

   A. The file has 3 repeated lines
   B. The header was counted as a line
   C. availableNow read the file in 2 batches
   D. Someone ran the job with an empty or new checkpoint, so it loaded every file again

### Answers

1. B. The failed run had already planned a batch with that file. With the file gone, the next run finished that batch empty and recorded the file as read, so the same name is skipped from then on. Fix the file where it is and run again, and the replayed batch reads the corrected file. A corrected copy under a new name is also loaded.
2. D. A checkpoint is the only record of what the stream has read, so a new or deleted one makes the next run read every file in the folder again, and every file shows double. Keep the checkpoint with the table, and if it is lost, rebuild the table from the files with a fresh checkpoint rather than appending to it.

## Questions

### How do I check that an ingestion job did not lose or duplicate data?

Count what came in and what was stored, per file. Read each source file as text, count its data lines, count the table's rows for the same file name, and fail the job when any file is missing from the table or the 2 counts differ. Keeping the file name on every row, from the _metadata column, is what makes the comparison possible.

### What should I do when a Spark stream fails on a bad file?

Fix the file where it is and run the stream again. The failed run had planned a batch with that file, and the restart replays that batch, so it reads the corrected file. Deleting the file instead lets the replayed batch finish without it, and the checkpoint then treats that file name as read.

## Sources

- [CSV files, options](https://spark.apache.org/docs/latest/sql-data-sources-csv.html)
- [Structured Streaming checkpoints](https://docs.databricks.com/aws/en/structured-streaming/checkpoints)
- [Configure schema inference and evolution in Auto Loader](https://docs.databricks.com/aws/en/ingestion/cloud-object-storage/auto-loader/schema)
- [File metadata column](https://docs.databricks.com/aws/en/ingestion/file-metadata-column)