Process a fixed time window from the run, never "the latest data"

A task that reads "whatever arrived since last time" or "the latest file" gives a different answer each time it runs. Airflow's docs are explicit: never read the latest available data in a task; read a specific partition, such as one keyed on the run's data_interval_start.

Each Airflow run has a data interval: the time range it covers. The logical date marks the start of that interval, not when the run executes. A daily run for 1 October usually starts after 1 October has ended, so all of that day's data is available.

Pass the interval into your queries and file paths, so the same run always touches the same slice of data.

from airflow.sdk import dag, task

@dag(schedule="@daily", catchup=False)
def orders_daily():
    @task
    def load(**context):
        start = context["data_interval_start"]
        end = context["data_interval_end"]
        load_orders_between(start, end)  # same window on every retry

    load()

orders_daily()