Build it: backfill a month with Airflow, find the failed day and rerun only that run

The module so far gave the shop's daily load a run date, made it safe to run twice, scheduled it in Lakeflow Jobs and in Airflow, and backfilled a month. This part is the job a data engineer gets in their first week: load September with Airflow while 1 day's file is missing, find that day, and fix it without touching the rest. It needs part 1's scripts, part 4's Airflow install and dags/shop_daily.py, in a fresh folder. To rebuild from nothing, run python make_days.py.

Build it: a month check

Each run's check_day.py checks the day that run loaded. It cannot see a day that no run loaded. A month check compares the whole table with the till's files, day by day. Save this as check_month.py:

"""Check a month of orders_daily against the till's files: every day of the month there, with every order once.
Exits 1 on a problem."""
import sys
from pyspark.sql import functions as F
from delta_session import delta_spark

month = sys.argv[1]  # such as 2026-09
spark = delta_spark("check_month")
days = spark.sql(
    f"SELECT explode(sequence(DATE'{month}-01', last_day(DATE'{month}-01'))) AS order_date"
)
files = (
    spark.read.option("header", True).schema("order_id INT")
    .csv(f"landing/orders/{month}-*.csv")
    .withColumn("order_date", F.to_date(F.regexp_extract(F.col("_metadata.file_name"), r"^(\d{4}-\d{2}-\d{2})\.csv$", 1)))
    .groupBy("order_date").agg(F.count("*").alias("in_file"))
)
table = spark.table("orders_daily").groupBy("order_date").agg(F.count("*").alias("in_table"))
compared = days.join(files, "order_date", "left").join(table, "order_date", "left").fillna(0, ["in_table"])
problems = compared.where("in_file IS NULL OR in_file != in_table").orderBy("order_date").collect()
for row in problems:
    if row.in_file is None:
        print(f"PROBLEM {row.order_date}: no file, {row.in_table} rows")
    else:
        print(f"PROBLEM {row.order_date}: {row.in_table} rows for the file's {row.in_file} orders")
total = compared.agg(F.sum("in_table")).first()[0]
if problems:
    print(f"PROBLEM {month}: {len(problems)} of {compared.count()} days do not match")
else:
    print(f"ok {month}: all {compared.count()} days match their files, {total} orders")
sys.exit(1 if problems else 0)

sequence and last_day make 1 row for every day of the month, so a day with no file and no rows still shows up. The file's day comes from its name, through the hidden _metadata column of a file read, which works the same on Databricks. A day is a problem when it has no file or its row count differs from its file's.

Now break the month the way it breaks in real life, with a file the till never sent:

python make_days.py
rm landing/orders/2026-09-20.csv

Turn the DAG on

Start Airflow in a second terminal with the same environment settings as part 4. airflow standalone sets up its database, creates a user and starts the scheduler and the UI, at localhost:8080, with the password in airflow/simple_auth_manager_passwords.json.generated. Then:

airflow dags unpause shop_daily

A new DAG is paused, and the runs of a backfill on a paused DAG wait in the queue. Turning it on also starts 1 run by itself. catchup=False stops Airflow creating a run for every date since start_date, but the scheduler still creates 1 for the latest scheduled time, 02:00 today. That run loads yesterday, which in this sample has no file, so it failed after its retries with [PATH_NOT_FOUND] Path does not exist. In a shop whose till sent yesterday's file, it would be the normal daily run.

Backfill September

The dry run lists the runs before anything is created:

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

Its table starts at 2026-09-02 02:00:00+00:00 and ends at 2026-10-01 02:00:00+00:00, 30 runs. The same command without --dry-run creates them. With --max-active-runs 1 they ran 1 at a time and took 22 minutes, 10 of them spent waiting for retries. airflow dags list-runs shop_daily then listed 29 backfill runs that succeeded, 1 that failed, the run for 2026-09-21 02:00, and today's failed run. airflow tasks states-for-dag-run shop_daily backfill__2026-09-21T02:00:00+00:00 gives the state of each task in the failed run: load_day failed and check_day upstream_failed.

The run for 21 September loads 20 September, whose file is gone. load_day was tried 3 times, 5 minutes apart, the 2 retries of part 4's default_args, and its log shows the same PATH_NOT_FOUND each time. check_day never ran: it is upstream_failed, because its default rule needs the load to succeed. The runs after it carried on, so a failed day does not stop a backfill. The month check finds it:

python check_month.py 2026-09

PROBLEM 2026-09-20: no file, 0 rows
PROBLEM 2026-09: 1 of 30 days do not match

It exits with 1, so as a task after the backfill it would fail where someone sees it.

Rerun only the failed day

The till sends the file late. make_days.py writes the same files again, so it puts back 20 September. Then the same backfill, told to reprocess failed runs:

python make_days.py
airflow backfill create --dag-id shop_daily --from-date 2026-09-02 --to-date 2026-10-02 --max-active-runs 1 --reprocess-behavior failed

Airflow ran only the run for 21 September. Its log for load_day has a 4th attempt, which printed run 2026-09-21 loaded 2026-09-20: 3 rows, orders_daily has 130, and check_day ran and printed ok 2026-09-20: 3 rows for the file's 3 orders, 0 missing. All 30 runs of the backfill then had succeeded. Airflow's backfill page speaks of a new run for each failed date; in Airflow 3.3.2 the list still had 31 runs, so it ran the failed run again under the same id. Today's failed run was outside the range and stayed failed. The month check passes:

ok 2026-09: all 30 days match their files, 130 orders

Nothing else changed. The 29 days that were right were not loaded again, and the 1 that ran again replaced only its own day, which is what replaceWhere in part 1 is for. On Databricks the same fix is Repair run on the failed run of a Lakeflow job, from part 2.

When you are done, pause the DAG with airflow dags pause shop_daily and stop airflow standalone with Ctrl+C, so its daily run does not keep failing on your computer.

What you built

• A daily load and its check that take their day from the run, and replace that day, so a rerun, a backfill or a correction is the same command.
• The same job in Lakeflow Jobs, with retries, a run if rule and repair, and as an Airflow DAG that can also start the Lakeflow job.
• A month check that finds the day no run loaded, and a backfill that reran only that day.

Module 11 tests this code before it ships: unit tests for the load, data contracts for its files, and the job deployed as a bundle.

ComparisonAirflow vs Dagster vs PrefectApache Airflow, Dagster and Prefect all run data pipelines on a schedule, retry what fails and show every run. Airflow schedules Dags of tasks and has managed…