The DAG That Runs Twice a Day and Reads Yesterday's Files

The DAG is not idempotent and it is not partitioned by the run's logical date. The task lists the whole prefix and loads whatever it finds, so any file present during two consecutive runs is loaded twice; that is the Tuesday duplicate. The missed Monday file is the opposite symptom of the same design: the file arrived at 07:00, after Monday's single run had already completed, and because catchup=False no further run was created to pick it up. Nothing in the DAG waits for the file to exist.

Changing the schedule to every 30 minutes makes the duplicate problem dramatically worse: there are now 48 chances per day to re-read the same prefix, and each run still loads everything it sees. retries=3 only helps when a task fails; a task that succeeds while loading stale data will never trigger a retry. Neither change fixes the root cause.

The correct direction is to make each run responsible for exactly one logical date's data, and to make the task wait for that data rather than hoping it is already there. Two complementary changes:

1. Use a sensor (for example an S3 key sensor with mode='reschedule' and a bounded timeout) that waits for the specific key for {{ ds }} before the load task runs. This handles the late 07:00 arrival without polling a worker slot all night.
2. Make the load idempotent - delete-and-insert or a MERGE keyed on the file's embedded date - so a re-run of the same logical date cannot duplicate rows.

With those in place the original @daily schedule is fine, and catchup=True becomes safe and useful for backfilling.

With catchup=True, the scheduler creates a DAG run for every interval between start_date and now, so enabling it for a 30-day-old start_date produces roughly 30 runs, one per logical date. Clearing historical task instances re-queues those tasks; the scheduler will then execute them, and because each run's templated ds differs, each run targets a different date.

The precondition is that the pipeline must already be idempotent and date-scoped. If the load task still scans the whole prefix, 30 backfill runs will each load every file they can see, producing a 30× duplication of the entire history. Backfill is only safe once the task reads exactly the data for its own logical date and writes with a deterministic key (delete-insert or MERGE).

Practical notes worth stating:

• Backfills compete with the live schedule for worker slots; consider pausing the DAG, backfilling, then unpausing, or using max_active_runs to throttle.
• The sensor's timeout must be long enough for historical files, or backfill runs will fail on files that were legitimately late.
• depends_on_past=True would serialize the 30 runs and is usually the wrong choice here; it turns a parallel backfill into a 30-step chain.
• Verify with a row count per partition after the backfill rather than trusting the green squares in the UI.

It is a poor trade even though it superficially addresses lateness. Problems:

• It always waits the full hour, even when the file arrived at 01:05, wasting a worker slot and delaying downstream work.
• It still is not sufficient: a 07:00 arrival is three hours late, so a one-hour sleep fails anyway, and lengthening the sleep makes the common case worse.
• It occupies a worker slot while sleeping. A sensor in mode='reschedule' releases the slot between checks; a sleeping task or a poke-mode sensor holds it. On a small executor this can starve other DAGs.
• It hides failure. If the file never arrives, the task proceeds and either crashes obscurely or silently loads nothing; a sensor with timeout fails loudly and is visible in the UI as a distinct state.
• It is untestable and opaque - the wait is buried in code rather than expressed as a dependency.

The sensor expresses the real business rule ('this run needs this file') as a first-class Airflow concept, is reschedulable, has a configurable timeout, and can be given poke_interval and mode appropriate to the expected delay. If the team wants a hard cutoff, a sensor with timeout=4*3600 and mode='reschedule' gives exactly that behaviour with far better operational visibility.