Backfill a data pipeline: a month of runs, a corrected day and parallel runs

Part 2 put the daily load in a Lakeflow job. The job only runs from now on; the shop wants its September in orders_daily too. Running a pipeline for past dates is a backfill, and this part does one locally, corrects a day, and then runs it in Lakeflow Jobs and Airflow. It uses part 1's scripts in a fresh folder. To rebuild from nothing, run python make_days.py.

A month, 1 run per day

An orchestrator splits a backfill's range into the job's own interval and starts 1 run for each, with its date. Locally, a loop does the same, in 1 Spark session so it does not start Spark 30 times. Save this as backfill.py:

"""Run the daily load and its check for every run date from the first to the last, in 1 Spark session,
as an orchestrator's backfill would run the job once per day."""
import sys
from datetime import date, timedelta
from delta_session import delta_spark
from load_day import load_day
from check_day import check_day

first, last = date.fromisoformat(sys.argv[1]), date.fromisoformat(sys.argv[2])
spark = delta_spark("backfill")
failed = []
run = first
while run <= last:
    load_day(spark, str(run))
    ok, message = check_day(spark, str(run))
    if not ok:
        failed.append(str(run))
        print(message)
    run += timedelta(days=1)
days = spark.table("orders_daily").select("order_date").distinct().count()
print(f"ran {(last - first).days + 1} runs, {len(failed)} failed; orders_daily has {days} days, {spark.table('orders_daily').count()} orders")

The runs for September's 30 days are dated 2 September to 1 October, because each loads the day before its date:

python backfill.py 2026-09-02 2026-10-01

ran 30 runs, 0 failed; orders_daily has 30 days, 130 orders

30 days and 130 orders, the 2001 to 2130 that make_days.py wrote. Because each run replaces only its own day, the same command can run again, or stop halfway and start over, and the table ends the same.

Correct 1 day

A correction is a backfill of 1 day. Say the till's file for 15 September was missing an order, sent late:

echo "2131,C02,19.99,2026-09-15T20:30:00" >> landing/orders/2026-09-15.csv
python load_day.py 2026-09-16
python check_day.py 2026-09-16
python history.py

run 2026-09-16 loaded 2026-09-15: 4 rows, orders_daily has 131
ok 2026-09-15: 4 rows for the file's 4 orders, 0 missing
+---------+-------------------------+------------+
|operation|predicate                |rows_written|
+---------+-------------------------+------------+
|WRITE    |order_date = '2026-09-15'|4           |
|WRITE    |order_date = '2026-09-30'|6           |
|WRITE    |order_date = '2026-09-29'|5           |
+---------+-------------------------+------------+

The run for 16 September replaced 15 September's 3 rows with 4, and the history names the day it replaced. The other 29 days were not touched. A load that took its day from the clock could not do this: run today, it would load yesterday.

Runs at the same time

30 runs one after another take a while, and orchestrators can run several at once. Runs that write the same Delta table at once can conflict, though. Save this as parallel.py, which starts 4 runs together:

"""Run 4 days' loads at the same time into orders_daily, as a backfill with 4 concurrent runs would."""
import re
from concurrent.futures import ThreadPoolExecutor
from delta_session import delta_spark
from load_day import load_day

spark = delta_spark("parallel")


def run(run_date: str) -> str:
    try:
        return load_day(spark, run_date)
    except Exception as error:
        # The error class in square brackets is the same in classic Spark and Spark Connect.
        found = re.search(r"\[([A-Z_.]+)\]", str(error))
        return f"run {run_date} failed: {found.group(1) if found else error}"


with ThreadPoolExecutor(4) as pool:
    for result in pool.map(run, ["2026-09-02", "2026-09-03", "2026-09-04", "2026-09-05"]):
        print(result)

python parallel.py printed:

run 2026-09-02 failed: DELTA_CONCURRENT_APPEND.WITHOUT_HINT
run 2026-09-03 loaded 2026-09-02: 5 rows, orders_daily has 131
run 2026-09-04 failed: DELTA_CONCURRENT_APPEND.WITHOUT_HINT
run 2026-09-05 failed: DELTA_CONCURRENT_APPEND.WITHOUT_HINT

DELTA_CONCURRENT_APPEND is the error class of Delta's ConcurrentAppendException, a write conflict. Which run wins changes from try to try; in every try, 1 of the 4 succeeded. Each replaceWhere reads the table to find the rows it replaces. orders_daily is not partitioned, so Delta counts each run as having read the whole table, and rows another run committed in the meantime as a change to what it read. It fails those commits rather than risk a wrong result. The failed runs changed nothing, because a Delta write commits whole or not at all, and running them again loads them.

2 ways out. Run the backfill 1 run at a time, which for a table this small costs little. Or partition the table by order_date, with .partitionBy("order_date") on its first write: each run then replaces only its own partition, and the same 4 runs into a partitioned copy of the table all succeeded, 3 times in 3. Partitioning by day makes many small files when days are small, the cost module 4 dealt with.

A backfill in Lakeflow Jobs

Open the job, click the arrow next to Run now and choose Run backfill. Set:

• Date and time range: 2 September to 1 October, the run dates, not the days loaded.
• Every: 1 day, which Databricks chooses from the job's schedule. The dialog shows how many runs the range makes; check it says 30. It will not start a range of more than 100 runs, so split a longer one into several backfills.
• Job parameters: override run_date with {{backfill.iso_date}}, the date at the start of each run's interval.

Backfill runs show in the job's runs with Backfill in their name. They go in parallel up to the job's maximum concurrent runs, which part 2 left at 1, the safe choice for this unpartitioned table. A backfill always runs every task of the job. If the job had a pipeline task, the pipeline would run as it is, with no parameters, and the runs would go one after another.

A backfill in Airflow

Airflow's backfill creates a run for each date of the schedule in a range, and gives each its date as {{ ds }}:

airflow backfill create --dag-id shop_daily --from-date 2026-09-02 --to-date 2026-10-02 --max-active-runs 1

The dates are read as midnight, so the runs at 02:00 fall from 2 September to 1 October, and the end date is the day after the last run. --reprocess-behavior decides what happens to dates that already have a run: none creates no new run for them, failed creates one where the run failed, and completed creates one for every finished run, failed or not. A date whose run is still queued or running gets no new run either way. --max-active-runs limits how many of the backfill's runs go at once, separately from the DAG's own max_active_runs, and --dry-run prints the dates without creating anything. In the UI the same backfill is under Trigger, Backfill. Part 4 writes the DAG that runs this, and part 5 runs this backfill.