Apache Airflow DAG for a daily Spark load: schedule, ds, retries and Databricks

Part 3 backfilled September with Lakeflow Jobs or a local loop. Apache Airflow is the orchestrator most data teams outside Databricks use, and it can start Databricks jobs too. This part runs the same 2 scripts from an Airflow DAG on your computer, then starts the Lakeflow job from one. It uses part 1's scripts; to rebuild from nothing, run python make_days.py.

Airflow in 1 paragraph

A pipeline in Airflow is a DAG, a Python file that names its tasks and the order they run in. The scheduler reads the DAG files, creates a DAG run for each scheduled time and queues its tasks once the tasks before them succeed. Every run has a logical date, which templates such as {{ ds }} give to the tasks. The UI and a database keep every run, every attempt of every task and its log. The Airflow, Dagster and Prefect comparison sets it beside 2 newer tools.

Install it

Airflow is installed with a constraints file that pins every library it needs to versions tested together, for 1 Airflow version and 1 Python version. In a new virtual environment with Python 3.12:

pip install "apache-airflow==3.3.2" --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-3.3.2/constraints-3.12.txt"
pip install pyspark==4.2.0 delta-spark==4.4.1

The second line puts Spark in the same environment, because the tasks run python load_day.py with the Python Airflow runs on. Airflow keeps its settings and database in AIRFLOW_HOME, ~/airflow unless you say otherwise. From the folder with part 1's scripts:

export AIRFLOW_HOME=$PWD/airflow AIRFLOW__CORE__DAGS_FOLDER=$PWD/dags AIRFLOW__CORE__LOAD_EXAMPLES=False
airflow db migrate

Any setting in airflow.cfg can be set as AIRFLOW__<SECTION>__<KEY>; these 2 point Airflow at a dags folder here and leave out the example DAGs.

The DAG

Save this as dags/shop_daily.py:

"""The shop's daily job: at 02:00 UTC, load the day before's orders file, then check it.
Each run passes its date, {{ ds }}: the run on 2 September loads 1 September."""
from datetime import datetime, timedelta
from pathlib import Path

from airflow.providers.standard.operators.bash import BashOperator
from airflow.sdk import dag

# The scripts and landing/ are in the folder above dags/.
FOLDER = str(Path(__file__).parent.parent)


@dag(
    schedule="0 2 * * *",
    start_date=datetime(2026, 9, 2),
    catchup=False,
    max_active_runs=1,
    default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},
)
def shop_daily():
    load = BashOperator(task_id="load_day", bash_command="python load_day.py {{ ds }}", cwd=FOLDER)
    check = BashOperator(task_id="check_day", bash_command="python check_day.py {{ ds }}", cwd=FOLDER)
    load >> check


shop_daily()

Line by line:

• schedule="0 2 * * *" is a standard 5-field cron, every day at 02:00, in Airflow's time zone, UTC unless set. The cron builder writes one from words.
• start_date is the first date the schedule counts from, and catchup=False, the default in Airflow 3, means turning the DAG on does not create runs for every date since then, only 1 for the latest scheduled time. Part 5 creates the rest on purpose, with a backfill.
• max_active_runs=1 lets 1 run go at a time, as max_concurrent_runs did in part 2.
• default_args applies to every task: up to 2 retries, 5 minutes apart.
• BashOperator runs a shell command. {{ ds }} is filled in with the run's date before it runs, and a command that exits with a code other than 0, as check_day.py does on a problem, fails the task.
• load >> check runs the check after the load, and only if the load succeeded, Airflow's default rule.

The airflow.providers.standard and airflow.sdk imports are Airflow 3's. A DAG written for Airflow 2 imports BashOperator from airflow.operators.bash and dag from airflow.decorators.

Test it

airflow dags test runs a DAG once for a date, in 1 process, with no scheduler:

airflow dags test shop_daily 2026-09-16

Among its many log lines are the scripts' own prints, and its last lines say how the run ended:

run 2026-09-16 loaded 2026-09-15: 3 rows, orders_daily has 3
ok 2026-09-15: 3 rows for the file's 3 orders, 0 missing
DagRun Finished: dag_id=shop_daily, logical_date=2026-09-16 00:00:00+00:00

The date given to dags test became the logical date, ds was 2026-09-16, and the load took the day before. It is the quickest way to try a DAG after an edit.

What ds means, in Airflow 3 and Airflow 2

In Airflow 3, a schedule given as a cron string uses CronTriggerTimetable: a run's logical date is the time it was scheduled for. The run at 02:00 on 2 September has the logical date 2026-09-02 02:00, and ds is 2026-09-02, which is why the scripts load the day before ds.

Airflow 2 used data intervals for the same string. The run at 02:00 on 2 September covered the interval from 02:00 on 1 September, and its logical date was the start of that interval, so ds was 2026-09-01. DAGs written then loaded ds itself. Upgrade one without changing anything and every run loads a day later than before, the day still in progress, and nothing fails. Airflow's upgrade guide says to choose before upgrading: set create_cron_data_intervals to True in the [scheduler] settings to keep the old dates, or change the tasks, as here.

To run a past day again, clear its task in the UI, or with airflow tasks clear shop_daily -s 2026-09-16 -e 2026-09-16. The scheduler runs it again in the same DAG run, with the same ds, so it loads the same day.

Airflow starting a Lakeflow job

When the work runs on Databricks, Airflow can start the Lakeflow job from part 2 and wait for it, instead of running Spark itself. It does this through the Databricks provider, an open source package from the Airflow project that Databricks does not support directly. Install it with the same constraints file, pip install apache-airflow-providers-databricks --constraint "<the same URL>", and save this as dags/shop_on_databricks.py:

"""Start the shop's Lakeflow job from Airflow at 02:00 UTC each day, with the run's date as its run_date."""
from datetime import datetime

from airflow.providers.databricks.operators.databricks import DatabricksRunNowOperator
from airflow.sdk import dag


@dag(schedule="0 2 * * *", start_date=datetime(2026, 9, 2), catchup=False, max_active_runs=1)
def shop_on_databricks():
    DatabricksRunNowOperator(
        task_id="run_shop_daily",
        job_name="shop_daily",  # the job from part 2; the connection databricks_default says which workspace
        job_parameters={"run_date": "{{ ds }}"},
    )


shop_on_databricks()

job_parameters overrides the job's own default for run_date, and is filled in from ds like a bash command. airflow tasks render shop_on_databricks run_shop_daily 2026-09-16 shows what the task would send, without sending it:

# property: job_parameters
# ----------------------------------------------------------
{'run_date': '2026-09-16'}

To send it, Airflow needs a connection named databricks_default with the workspace URL as its host and a personal access token as its password, set in the UI under Admin, Connections, or as an AIRFLOW_CONN_DATABRICKS_DEFAULT environment variable. Keep the token out of the DAG file. The operator waits for the Databricks run to end and fails if it fails, so Airflow's retries start the job again.

Run the schedule in 1 place. If Airflow starts the job, set the job's own schedule to pause_status: PAUSED, or every day runs twice.

Which orchestrator

If the whole pipeline runs on Databricks, Lakeflow Jobs does all of this with nothing more to run, and repairs and backfills from the same page. Airflow earns its place when a pipeline spans systems outside Databricks, such as a file transfer, a dbt project and an export to a partner's system, or when a team already runs it. Either way, the 2 rules of part 1 are what make it safe: the day comes from the run's date, and loading it again replaces it.