Apache Airflow Fundamentals: DAGs, Tasks, and Scheduling

A workflow has definitions and executions

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.

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.

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 check rejects negative totals and counts, but 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. Run the setup before each example. Only the setup belongs in a DAG file; the following assertions are separate local tests. The fixed three-event fixture and tiny XCom summary demonstrate boundaries, not a production ingestion or publication task.

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.

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

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