Data pipeline orchestration: run dates, idempotent loads and retries

Module 9 kept a table up to date as events arrived. Most of a shop's data still arrives in batches, a file a night, and something has to start each load at the right time, in the right order, and again when it fails. This module is about that something. This part needs module 4's delta_session.py, the helper that starts a local Spark session with Delta Lake.

What an orchestrator does

An orchestratorOrchestrationRunning the steps of data pipelines in the right order and at the right time, retrying failures and alerting someone when a step breaks. runs a pipeline's tasks for you. It starts a run on a schedule or an event, runs the tasks in order, so a check waits for the load it checks, retries a task that fails, keeps a record of every run and its result, and can run the pipeline again for past dates, which is a backfill. On Databricks the orchestrator is Lakeflow Jobs, which part 2 covers. Apache Airflow is the most common one outside it, in part 4, and it can also start Databricks jobs.

None of that makes a pipeline correct. A retry runs a task again after it may have written something, and a backfill runs it for days long gone. Both are only safe if running a task twice for the same day leaves the same rows, which is called being idempotentIdempotent Pipelines and Exactly-OnceA pipeline step is idempotent when running it twice gives the same result as running it once. That is what makes retries safe, and it is how systems that deliver a message more than once still count it exactly once.. This part builds a daily load that is.

A file a night

The shop's till writes each day's orders to a file after midnight. Save this as make_days.py; it writes the 30 days of September 2026:

"""Write 1 file of orders for each day of September 2026 into landing/orders/, as the shop's till does each night."""
import csv
import random
from datetime import date, datetime
from pathlib import Path

random.seed(10)
folder = Path("landing/orders")
folder.mkdir(parents=True, exist_ok=True)
order_id = 2001
for day in range(1, 31):
    when = date(2026, 9, day)
    with open(folder / f"{when}.csv", "w", newline="") as file:
        writer = csv.writer(file)
        writer.writerow(["order_id", "customer_id", "amount", "ordered_at"])
        times = sorted(datetime(2026, 9, day, random.randint(8, 21), random.choice([0, 15, 30, 45])) for _ in range(random.randint(3, 6)))
        for at in times:
            writer.writerow([order_id, f"C0{random.randint(1, 5)}", f"{random.randint(800, 6000) / 100:.2f}", at.isoformat()])
            order_id += 1
print("wrote", len(list(folder.glob("*.csv"))), "days, orders", 2001, "to", order_id - 1)

python make_days.py prints wrote 30 days, orders 2001 to 2130. The first file, landing/orders/2026-09-01.csv, holds 3 orders:

order_id,customer_id,amount,ordered_at
2001,C04,30.73,2026-09-01T11:45:00
2002,C02,10.81,2026-09-01T14:45:00
2003,C05,48.14,2026-09-01T17:00:00

The run's date, not the clock's

The job runs at 02:00 and loads the day before, because a day's file is complete only after midnight. The tempting way to find that day is date.today() - timedelta(days=1). It works every night the job runs on time, and that is the trouble: the clock says when a run started, not which day it is for. Retry a run after midnight, rerun a broken day a week later or backfill a month, and the run loads the wrong day without failing.

Instead, every run is given a date by the orchestrator, and the script loads the day before that date. Airflow passes it as {{ ds }}; a Lakeflow job passes {{job.trigger.time.iso_date}}, the day the run was triggered. Press the controls to see the difference:

day = date.today() - timedelta(days=1)

Airflow keeps a run's date when you run it again. A Lakeflow job takes the date as a parameter you can set, so part 2 runs a past day by giving it that day.

A load that is safe to run twice

Save this as load_day.py:

"""The shop's daily load. A run names the day it runs, and loads the day before: the till's file for a day
is complete only after midnight. Safe to run again: it replaces that day's rows and leaves the rest alone.
--append shows what a plain append does instead."""
import sys
from datetime import date, timedelta
from pyspark.sql import SparkSession, functions as F


def load_day(spark: SparkSession, run_date: str, append: bool = False) -> str:
    day = str(date.fromisoformat(run_date) - timedelta(days=1))  # the run on 2026-09-02 loads 2026-09-01
    orders = (
        spark.read.option("header", True)
        .schema("order_id INT, customer_id STRING, amount DECIMAL(10,2), ordered_at TIMESTAMP")
        .csv(f"landing/orders/{day}.csv")
        .withColumn("order_date", F.lit(day).cast("date"))
    )
    writer = orders.write.format("delta")
    if append:
        writer.mode("append").saveAsTable("orders_daily")
    else:
        # Replace this day's rows, and only this day's, in 1 transaction.
        writer.mode("overwrite").option("replaceWhere", f"order_date = '{day}'").saveAsTable("orders_daily")
    table = spark.table("orders_daily")
    return f"run {run_date} loaded {day}: {table.where(F.col('order_date') == day).count()} rows, orders_daily has {table.count()}"


if __name__ == "__main__":
    from delta_session import delta_spark

    print(load_day(delta_spark("load_day"), sys.argv[1], append="--append" in sys.argv))

mode("overwrite") with replaceWhere deletes the rows that match the condition and writes the new ones in 1 Delta transaction, so a reader sees the day before or after, never half of it. The condition can be any expression over the table's columns; the table does not need to be partitioned by order_date. Delta also checks that every row written matches the condition, and fails the write with DELTA_REPLACE_WHERE_MISMATCH if one does not, so a bug cannot overwrite 1 day with another's rows.

The load is a function so part 3's backfill can call it for many days in 1 session. Load the runs for 2 and 3 September, then the run for 2 September again:

python load_day.py 2026-09-02
python load_day.py 2026-09-03
python load_day.py 2026-09-02

run 2026-09-02 loaded 2026-09-01: 3 rows, orders_daily has 3
run 2026-09-03 loaded 2026-09-02: 5 rows, orders_daily has 8
run 2026-09-02 loaded 2026-09-01: 3 rows, orders_daily has 8

Running a day again replaced its 3 rows. The table still has 8.

Check the day, and break it

A load that exits 0 has only proved it wrote something. Save this check as check_day.py; it compares the day in the table with its file:

"""Check the day a run loaded against its file: every order, once. Exits 1 on a problem."""
import sys
from datetime import date, timedelta
from pyspark.sql import SparkSession, functions as F


def check_day(spark: SparkSession, run_date: str) -> tuple[bool, str]:
    day = str(date.fromisoformat(run_date) - timedelta(days=1))
    file = spark.read.option("header", True).schema("order_id INT").csv(f"landing/orders/{day}.csv")
    loaded = spark.table("orders_daily").where(F.col("order_date") == day)
    expected, rows = file.count(), loaded.count()
    missing = file.join(loaded, "order_id", "left_anti").count()
    ok = rows == expected and missing == 0
    return ok, f"{'ok' if ok else 'PROBLEM'} {day}: {rows} rows for the file's {expected} orders, {missing} missing"


if __name__ == "__main__":
    from delta_session import delta_spark

    ok, message = check_day(delta_spark("check_day"), sys.argv[1])
    print(message)
    sys.exit(0 if ok else 1)

Its exit code is what an orchestrator reads: 1 fails the task. Now run the day again the way an INSERT would, check it, then load it properly and check again:

python check_day.py 2026-09-02
python load_day.py 2026-09-02 --append
python check_day.py 2026-09-02
python load_day.py 2026-09-02
python check_day.py 2026-09-02

ok 2026-09-01: 3 rows for the file's 3 orders, 0 missing
run 2026-09-02 loaded 2026-09-01: 6 rows, orders_daily has 11
PROBLEM 2026-09-01: 6 rows for the file's 3 orders, 0 missing
run 2026-09-02 loaded 2026-09-01: 3 rows, orders_daily has 8
ok 2026-09-01: 3 rows for the file's 3 orders, 0 missing

The append succeeded and doubled the day; only the check, exiting 1, said so. The proper load fixed it, because replacing a day does not care what was there. The table's history shows the 3 writes. Save this as history.py:

"""Show the last 3 changes to orders_daily: what each did, and to which rows."""
from delta_session import delta_spark

spark = delta_spark("history")
(
    spark.sql("DESCRIBE HISTORY orders_daily LIMIT 3")
    .selectExpr("operation", "operationParameters.predicate AS predicate", "operationMetrics.numOutputRows AS rows_written")
    .show(truncate=False)
)

python history.py, newest first:

+---------+-------------------------+------------+
|operation|predicate                |rows_written|
+---------+-------------------------+------------+
|WRITE    |order_date = '2026-09-01'|3           |
|WRITE    |NULL                     |3           |
|WRITE    |order_date = '2026-09-01'|3           |
+---------+-------------------------+------------+

The append is the write with no predicate: it replaced nothing.

Why retries need this

Airflow's best practices say to treat a task like a database transaction: it should produce the same outcome on every re-run, so use an upsert instead of an INSERT, read a fixed partition instead of the latest data, and never use now() for the work itself. A task can fail after its write has committed, when the check after it fails or the machine is lost before the orchestrator hears back, and the retry then runs the whole task again. With the run's date and replaceWhere, that second run changes nothing.

1 case to know: if a day's file has a header and no orders, the load replaces that day's rows with none, which is right for a day the shop was shut. A missing file fails the read, so nothing is written. Part 5 relies on that.