Build an incremental ingestion job in PySpark that loads every file 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 and make_files.py from part 1.

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:

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

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:

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:

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 driftSchema DriftWhen the shape of incoming data changes without warning, such as a renamed or new column, and breaks or silently corrupts the pipelines that read it. 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:

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 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.

ToolCron expression builderA schedule in words in; the cron, Quartz and Snowflake expressions, the code and the next runs out.