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

An Airflow DAG is a Python file that names a pipeline's tasks, their order and its schedule. For a daily load, give it a cron schedule, pass each task the run's date with the template {{ ds }}, and set retries in default_args. In Airflow 3, a bare cron schedule makes ds the date the run was scheduled for, so the run at 02:00 on 2 September gets 2026-09-02; Airflow 2 gave it the start of the interval, 2026-09-01. To run the work on Databricks, a DAG can start a Lakeflow job with DatabricksRunNowOperator and pass the date as a job parameter.

- Page: https://tecyfy.com/data-engineering/course/orchestration/airflow
- Course: Data engineering course (https://tecyfy.com/data-engineering/course), Module 10: Orchestration, Apache Airflow, the same job as a DAG
- Parts of this module: Run dates, safe reruns and retries (https://tecyfy.com/data-engineering/course/orchestration); Lakeflow Jobs, a daily job on Databricks (https://tecyfy.com/data-engineering/course/orchestration/lakeflow-jobs); Backfills and corrections (https://tecyfy.com/data-engineering/course/orchestration/backfill); Apache Airflow, the same job as a DAG (https://tecyfy.com/data-engineering/course/orchestration/airflow); Build it: backfill a month, rerun the failed day (https://tecyfy.com/data-engineering/course/orchestration/build-it)

At the end of this part you can:

- Write an Airflow 3 DAG that runs a daily load and its check with the run's date, and test it without a scheduler
- Say what ds is for a cron schedule in Airflow 3, and why a DAG written for Airflow 2 can load a different day after an upgrade
- Start a Lakeflow job from Airflow with DatabricksRunNowOperator, and decide when Airflow is worth it over Lakeflow Jobs alone

[Part 3](/data-engineering/course/orchestration/backfill) 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](/data-engineering/airflow-vs-dagster-vs-prefect) 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:

```bash
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:

```bash
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`:

```python
"""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](/data-engineering/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:

```bash
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:

```text
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`:

```python
"""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:

```text
# 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](/data-engineering/course/orchestration) are what make it safe: the day comes from the run's date, and loading it again replaces it.

## Check yourself

1. A DAG with schedule="0 2 * * *" ran on Airflow 2 with tasks that loaded the day {{ ds }} names. After an upgrade to Airflow 3 with default settings, which day does the run at 02:00 on 2 September load?

   A. 1 September, as before
   B. 2 September, the day it runs, whose file is not finished
   C. Nothing, because ds was removed in Airflow 3
   D. 31 August

### Answers

1. B. Airflow 2 used a data interval timetable for a cron string, so the run at 02:00 on 2 September had the logical date 1 September, the start of its interval. Airflow 3 uses CronTriggerTimetable by default, so the logical date, and ds with it, is 2 September. Set create_cron_data_intervals to True before upgrading to keep the old dates, or change the tasks to load the day before ds, as this module's scripts do.

## Questions

### What does {{ ds }} mean in Airflow 3?

The Dag run's logical date as YYYY-MM-DD. With a bare cron string as the schedule, Airflow 3 uses CronTriggerTimetable, so the logical date is the time the run was scheduled for, and the run at 02:00 on 2 September has ds 2026-09-02. A backfill gives each run the date it would have been scheduled for, and clearing a task runs it again with its run's date.

### Should I use Airflow or Lakeflow Jobs for Databricks pipelines?

If the work runs on Databricks, Lakeflow Jobs does it with no extra system to run, and can be repaired and backfilled from the same place. Airflow is worth it when a pipeline spans systems outside Databricks, or a team already runs it; it can start Lakeflow jobs with DatabricksRunNowOperator and wait for them. Run each schedule in 1 place only, so a job is not started twice.

## Sources

- [Quick Start](https://airflow.apache.org/docs/apache-airflow/stable/start.html)
- [Upgrading to Airflow 3](https://airflow.apache.org/docs/apache-airflow/stable/installation/upgrading_to_airflow3.html)
- [Timetables](https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/timetable.html)
- [Dag Runs](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/dag-run.html)
- [Templates reference](https://airflow.apache.org/docs/apache-airflow/stable/templates-ref.html)
- [Testing a DAG](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/debug.html)
- [DatabricksRunNowOperator](https://airflow.apache.org/docs/apache-airflow-providers-databricks/stable/operators/run_now.html)
- [Databricks Connection](https://airflow.apache.org/docs/apache-airflow-providers-databricks/stable/connections/databricks.html)
- [Orchestrate Lakeflow Jobs with Apache Airflow](https://docs.databricks.com/aws/en/jobs/how-to/use-airflow-with-jobs)