Apache Airflow Fundamentals

Reading note: the practical examples and exercises are optional. Focus on the concepts, architecture, assumptions, and design decisions; no execution or project submission is required for this reading path.

A workflow has definitions and executions

This tutorial assumes you can read Python functions and basic SQL. You will build a two-task dependency graph, identify the time range a scheduled run should process, and test what happens when a destination write is repeated. The examples use Airflow 3.1.0 and Python 3.12; the API imports are specific to that environment.

A daily report may require a calculation followed by a quality check. Apache Airflow records that dependency, decides when a run is eligible, and tracks the attempts that perform the work. It can execute Python or submit work to another system. Large transformations still need appropriate compute and storage; an arrow between tasks neither moves a file nor validates its contents.

An Airflow DAG and DAG Run have different roles. The DAG is a directed acyclic graph of task definitions: dependencies have a direction and cannot form a cycle. A DAG Run is one execution of that definition. The same calculate task can have separate instances in Monday’s and Tuesday’s runs, and those runs may overlap unless configured otherwise.

An operator supplies a type of work; a TaskFlow-decorated Python function also creates a task. Calling that decorated function while building a DAG creates a task reference, not an immediate calculation. Passing the reference to another decorated task establishes a dependency and an XCom input. Independent tasks may run concurrently when capacity and their rules permit.

From parsed code to a worker

The DAG processor loads definitions. The scheduler evaluates runs, dependencies, and available capacity, and the configured executor arranges execution. A worker performs the task. Metadata tracks run and task state, while the UI, logs, and APIs help operators inspect it. Do not confuse the metadata database with the analytical destination that holds the report.

DAG files are parsed repeatedly. Keep top-level code limited to constructing the workflow; move source queries and heavy work inside tasks. Two tasks may execute on different machines. A local filename from one worker is therefore not a reliable transfer contract for the next.

Airflow XCom carries small task-to-task values. The lab returns a tiny summary; a real large result should normally stay in shared storage and pass a versioned URI or manifest reference. A reference is useful only if downstream workers have access and the referenced version remains available. Configure connections and secret backends for credentials rather than embedding them in DAG code or XCom payloads.

A calendar interval is not the start of an attempt

Data Interval and Logical Date describe scheduling context, not a worker’s clock. In this lab, an explicit CronDataIntervalTimetable uses UTC midnight boundaries. The interval [2024-10-01 00:00, 2024-10-02 00:00) becomes eligible at its end. Its logical date is the interval start. A worker starting late on October 3 must still use the October 1 interval.

The left boundary is included and the right boundary excluded. The event at October 2 midnight belongs to the next interval, so it is not counted twice. start_date sets the earliest interval boundary in this example; it does not promise a worker starts at that instant. Eligibility also does not prove all source events have arrived.

Timetable choice matters. Trigger-based, asset-driven, manual, and custom schedules need not represent the same daily interval or logical-date convention. Validate the context your workflow receives rather than assuming every run has yesterday’s range. Use a timezone-aware fixed start date; local calendar days around daylight-saving changes need not contain 24 elapsed hours.

Missing runs, retries, and historical work

For an interval schedule, catchup=True lets the scheduler create eligible historical runs that are missing. catchup=False avoids automatically filling that backlog; it does not forbid an explicit historical run or make now() a safe source filter. Set the policy explicitly and write the transformation for its interval.

For example, suppose this UTC daily DAG has no previous runs and is first enabled at noon on October 4, 2024. With catchup=False, its latest completed interval is October 3–4; it does not automatically create the October 1–2 and October 2–3 runs. With catchup=True, all three intervals are eligible. The scheduler can create runs for them, while max_active_runs=1 limits how many are active together. The still-open October 4–5 interval must wait until its end. See the Airflow 3.1.0 timetable comparison for why trigger timetables behave differently.

A Data Backfill deliberately selects historical work. Define the range, reprocessing policy, input and code versions, and limits before starting it. Clearing a task for re-execution is another operation: inspect which upstream and downstream instances will be affected. Neither action restores source history that no longer exists.

Task Instance and Retry separate an execution from its attempts. An ordinary failing task with retries=2 can make an initial attempt and up to two retries. A retry repeats work; it does not undo a database commit or an external notification from an earlier attempt. Use stable operation identity, transactions, or version-aware replacement at the destination, and retain evidence for ambiguous outcomes.

A retryable failure may leave a task up_for_retry until another attempt is allowed; a final failure is failed. A downstream task blocked by failed upstream work may become upstream_failed without running its own code. skipped represents omitted work, such as a branch not selected. Inspect the task instance and attempt logs rather than treating every non-success state as the same error.

The lab commits a summary and then raises an acknowledgement error. Repeating the same fixed summary overwrites the same interval key, leaving one result. This demonstrates a destination rule, not Airflow’s retry scheduler. A changed input version, concurrent writer, deletion, or a second external system requires additional rules.

Make the success path mean something

A Trigger Rule decides when upstream states satisfy a task’s dependency condition. The default all_success waits for upstream success. all_done allows a task to run after upstream tasks reach terminal states, including failures; it is useful for cleanup but does not mean the data passed its checks.

A leaf task has no downstream tasks. DAG Run status uses leaf task states. A successful all_done cleanup leaf can hide an earlier failure in the run’s overall status. Keep the data-validation path explicit and verify terminal-state behavior when adding cleanup or branching. A green graph records configured task outcomes, not an automatic proof of data completeness.

The sample assumes nonnegative event values, so its check rejects negative totals and counts. Refunds or signed adjustments would need a different business rule. Even here, a missing event can still produce a plausible nonnegative total. Production checks should connect to expected deliveries, row grain, accepted/rejected counts, and the intended published version. Alert the accountable owner with the DAG ID, run ID, task, attempt, interval, error, and destination evidence; do not expose secrets in logs.

Limit load and test each boundary

max_active_runs limits simultaneous runs of this DAG; it is not a global database connection cap. Pools limit participating tasks against a shared resource, while task and executor limits constrain other dimensions. Size them together. Retry delay and execution timeout control waiting between retries and task execution duration, not a complete business freshness guarantee.

Before enabling a DAG, test import and graph structure, time boundaries, transformation results, destination retry behavior, and then execution in the intended deployment. Inspect queued and running states, retry logs, permissions, and output visibility there. The following local checks cover the first boundaries. They do not start a scheduler, run workers, transfer XCom through a backend, or exercise live alerts.

Lab: inspect a DAG and its interval contract

Use Python 3.12 with apache-airflow==3.1.0, installed in an isolated environment with the official constraints-3.1.0/constraints-3.12.txt file. Set AIRFLOW_HOME to a disposable directory before importing Airflow. This is a version-pinned learning environment, not a deployment recommendation. Save the first Python block as daily_demo.py. Save the three solution blocks, in their displayed order, in check_daily_demo.py, preceded by from daily_demo import workflow, summarize, validate, pendulum, TimeRestriction, sqlite3. Keep both files in the same lab directory. Only daily_demo.py is a DAG definition; keep the assertion script outside a deployed DAG directory. The fixed three-event fixture and tiny XCom summary demonstrate boundaries, not a production ingestion or publication task.

In a new lab directory on macOS or Linux, run these commands. They create a local virtual environment and install dependencies from the network, following the Airflow 3.1.0 installation guide.

python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install "apache-airflow==3.1.0" --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-3.1.0/constraints-3.12.txt"
export AIRFLOW_HOME="$PWD/.airflow-lab"
python -m pip check

from datetime import timedelta
import sqlite3
import pendulum
from airflow.sdk import dag, task, get_current_context
from airflow.timetables.interval import CronDataIntervalTimetable
from airflow.timetables.base import TimeRestriction

EVENTS = [
    ("2024-10-01T00:00:00Z", 0),
    ("2024-10-01T12:00:00Z", 10),
    ("2024-10-02T00:00:00Z", 7),
]

def summarize(start, end):
    if start >= end:
        raise ValueError("empty or reversed interval")
    values = [v for timestamp, v in EVENTS if start <= pendulum.parse(timestamp) < end]
    return {"start": start.isoformat(), "end": end.isoformat(),
            "count": len(values), "total": sum(values)}

def validate(summary):
    if summary["count"] < 0 or summary["total"] < 0:
        raise ValueError("invalid summary")
    return summary

@dag(
    dag_id="daily_demo",
    start_date=pendulum.datetime(2024, 10, 1, tz="UTC"),
    schedule=CronDataIntervalTimetable("0 0 * * *", timezone="UTC"),
    catchup=False,
    max_active_runs=1,
    default_args={"retries": 2, "retry_delay": timedelta(minutes=1),
                  "execution_timeout": timedelta(minutes=5)},
)
def daily_demo():
    @task
    def calculate():
        context = get_current_context()
        return summarize(context["data_interval_start"], context["data_interval_end"])

    @task
    def check_quality(summary):
        return validate(summary)

    check_quality(calculate())

workflow = daily_demo()

Inspect the two task definitions and their dependency, retry, and catchup settings.

The fixture contains only October 1–2, 2024 events. A current scheduled interval will therefore produce count=0 and total=0, which this minimal validator accepts. To reproduce the results below, use the fixed historical intervals in the tests. Changing start_date alone with catchup=False does not replay those events.

assert set(workflow.task_ids) == {"calculate", "check_quality"}
assert workflow.get_task("check_quality").upstream_task_ids == {"calculate"}
assert workflow.get_task("calculate").upstream_task_ids == set()
assert workflow.get_task("calculate").retries == 2
assert workflow.get_task("check_quality").trigger_rule == "all_success"
assert workflow.catchup is False and workflow.max_active_runs == 1
print("TaskFlow graph: calculate -> check_quality")
print("retry allowance:", workflow.get_task("calculate").retries)
print("catchup and max_active_runs:", workflow.catchup, workflow.max_active_runs)
# TaskFlow graph: calculate -> check_quality
# retry allowance: 2
# catchup and max_active_runs: False 1

Ask the timetable for two adjacent intervals. The test deliberately uses catchup=True to enumerate from a fixed historical boundary; it does not change the DAG’s catchup=False setting or simulate today’s scheduler backlog.

start = pendulum.datetime(2024, 10, 1, tz="UTC")
restriction = TimeRestriction(earliest=start, latest=None, catchup=True)
first = workflow.timetable.next_dagrun_info(last_automated_data_interval=None, restriction=restriction)
second = workflow.timetable.next_dagrun_info(last_automated_data_interval=first.data_interval, restriction=restriction)
assert first.data_interval.start == start
assert first.data_interval.end == start.add(days=1)
assert first.run_after == first.data_interval.end
assert second.data_interval.start == first.data_interval.end
assert second.data_interval.end == start.add(days=2)
one = summarize(first.data_interval.start, first.data_interval.end)
two = summarize(second.data_interval.start, second.data_interval.end)
assert (one["count"], one["total"]) == (2, 10)
assert (two["count"], two["total"]) == (1, 7)
print("first interval:", first.data_interval.start.to_date_string(), first.data_interval.end.to_date_string())
print("eligible after:", first.run_after.to_iso8601_string())
print("adjacent summaries:", (one["count"], one["total"]), (two["count"], two["total"]))
# first interval: 2024-10-01 2024-10-02
# eligible after: 2024-10-02T00:00:00Z
# adjacent summaries: (2, 10) (1, 7)

Commit once, lose the acknowledgement, then repeat the same fixed result. The SQLite database is in memory and is closed afterward.

def commit_summary(db, summary, fail_after_commit=False):
    with db:
        db.execute("INSERT INTO daily VALUES (?, ?, ?, ?) ON CONFLICT(start, end) DO UPDATE SET n=excluded.n, total=excluded.total",
                   (summary["start"], summary["end"], summary["count"], summary["total"]))
    if fail_after_commit:
        raise ConnectionError("acknowledgement lost after commit")

start = pendulum.datetime(2024, 10, 1, tz="UTC")
summary = validate(summarize(start, start.add(days=1)))
db = sqlite3.connect(":memory:")
try:
    db.execute("CREATE TABLE daily(start TEXT, end TEXT, n INTEGER, total INTEGER, PRIMARY KEY(start,end))")
    try:
        commit_summary(db, summary, fail_after_commit=True)
    except ConnectionError as error:
        assert str(error) == "acknowledgement lost after commit"
    else:
        raise AssertionError("expected failure missing")
    commit_summary(db, summary)
    assert db.execute("SELECT n,total FROM daily").fetchall() == [(2,10)]
    bad = dict(summary, total=-1)
    try:
        validate(bad)
    except ValueError:
        pass
    else:
        raise AssertionError("quality check accepted negative total")
    print("commit then retry leaves one result:", (2,10))
    print("negative total rejected:", True)
finally:
    db.close()
# commit then retry leaves one result: (2, 10)
# negative total rejected: True

After saving all three solutions, run python check_daily_demo.py in the activated environment. Successful assertions finish without an exception and print the eight result lines shown in the solutions; Airflow may also emit logs. This executes local checks, not task instances. When finished, run deactivate. The virtual environment and .airflow-lab directory contain disposable lab files; retain them if you want to repeat the checks.

References: TaskFlow, DAG runs and status, Timetables, XCom, Best practices, Installation constraints.