Apache Airflow Architecture and DAGs

Data Engineering & Modern Data Platforms · Lesson 20 · Data Ingestion, ETL and Orchestration

Lesson Overview

The graph from Data Pipeline Orchestration Fundamentals is a design. Apache Airflow 2.x is one system that can run it. Airflow does not parse cents and it does not decide that country is a type-2 attribute. It parses Python files that declare tasks, stores runs in a metadata database, and asks an executor to start those tasks on workers. When a worker dies after the merge commits, Airflow may run the task again. The merge from Data Cleansing and Transformation is what makes that safe.

This lesson is about that machinery and about a complete DAG for the TechLearningPro daily warehouse load, aimed at Airflow 2.9. The same DAG shape works on 2.4 and later if you stay with the schedule argument, the TaskFlow API, and explicit data intervals. Dataset-aware scheduling, pools, sensors, and backfill operations are the next lesson. Here you need to see who parses the file, who creates the run, who executes the task, what a logical date is inside Airflow, how a dependency is written, and why an XCom is a bad place to put a DataFrame.

A DAG file is imported by a process that is not your task. Anything expensive or non-deterministic at import time runs on every parse, on the scheduler host, often every thirty seconds. The file should build a graph and return. The work belongs in task bodies that receive data_interval_start and data_interval_end.

Learning Objectives

By the end of this lesson, you should be able to:

1. Name the Airflow 2.x processes and what each is allowed to do.
2. Explain how a Python file becomes a serialized DAG in the metadata database.
3. Distinguish a DAG, a DagRun, a task, and a task instance.
4. Set schedule, start_date, and catchup without launching a surprise backfill.
5. Use data_interval_start and data_interval_end as the business window.
6. Write dependencies with TaskFlow and with classic operators.
7. Pass small values through XCom and keep row payloads in object storage or the warehouse.
8. Choose an executor at the level of where task processes run.
9. Keep credentials in Connections, not in the DAG source.
10. Read a task's failure as an attempt with a log, not as a mysterious red box.
11. Map the lesson 19 graph onto one DAG with a single publish task.
12. Say what this DAG will hand to Spark later: an interval and a task boundary, not a rewrite of the SQL.

Prerequisites

• Data Pipeline Orchestration Fundamentals: graph, interval, retries, pools, freshness. This lesson names the Airflow objects that implement them.
• Data Cleansing and Transformation: the functions the tasks call.
• Dimensional Modeling and Data Warehousing: load order.
• Python functions, context managers, and the idea of a database row as state.
• Airflow 2.9 vocabulary. Where a behavior arrived in an earlier 2.x minor release, the lesson says so.

Detailed Core Concepts

What Airflow is

Airflow is a workflow scheduler. You describe work in Python. A scheduler decides which task instances are eligible. An executor starts them. A web server shows state and logs. A metadata database remembers DAG structure, runs, attempts, and small XCom values. Airflow is not a data processing engine, not a warehouse, and not a streaming system. A task can submit Spark, run SQL, or call Python. The heavy bytes stay in the warehouse or the lake.

The project you deploy is a set of DAG files plus plugins and provider packages, on a shared Airflow deployment with one metadata database. TechLearningPro runs one production deployment. Development uses a separate metadata database. They do not share a database, because a dev DAG with the wrong connection id will then write production.

Processes in an Airflow 2.9 deployment

The web server does not decide to run a task. Killing the web server does not stop the scheduler. Stopping the scheduler stops new work from being queued. Stopping the workers stops running tasks. Stopping the database stops everything that needs state, which is everything.

Providers are separate packages. PostgreSQL operators ship in apache-airflow-providers-postgres and, for the generic SQL operator used below, apache-airflow-providers-common-sql. Core Airflow does not vendor every warehouse client. Pin provider versions beside the Airflow version. An unpinned install on a worker that drifts from the scheduler is a common way to get "works in the UI, fails on the worker."

DAG, DagRun, task, task instance

A DAG is the definition: id, schedule, default arguments, and tasks. It is code.

A DagRun is one execution of that definition for one data interval. The run that owns 1 March 2026 is a different row from the run that owns 2 March 2026.

A task is a node in the definition, such as load_fact_order_line.

A task instance is that task inside one DagRun. It has state and attempts. The 1 March instance and the 2 March instance of the same task are different rows. Retries are attempts of one task instance.

People say "the task failed" and point at the grid. The precise statement is "attempt 2 of task instance load_fact_order_line for the DagRun with data interval ending 2026-03-02 failed." Logs are per attempt. You will waste an hour reading attempt 1 when attempt 2 is the one that matters.

Parsing must be cheap and side-effect free

The DAG processor imports your module repeatedly. Top-level code runs on import.

Allowed at import: constructing DAG and operator objects, declaring dependencies, reading small static config.

Forbidden at import: querying the warehouse, listing a bucket with thousands of keys, calling an external API, opening a connection, generating a random run id, or branching on the current time in a way that changes the task ids. Task ids that change every parse look like deleted tasks and new tasks. The grid becomes noise, and historical runs no longer match the definition.

Database connections at import also mean the scheduler cannot parse when the warehouse is down, so unrelated scheduling in that file stalls behind an import error. Parse failure of one file should not be caused by a source system.

pythonCopy
# Wrong: runs on every parse, on the scheduler.
rows = warehouse_hook().get_records("SELECT id FROM analytics.dim_course")

# Right: a task lists what it needs, at run time, for its own interval.

Dynamic task ids that depend on a query belong in a mapped task at runtime, which lesson 21 covers, not in a list comprehension at import that hits production.

Schedule, start date, catchup

From Airflow 2.4, schedule is the argument. It accepts a cron string, a timedelta, a timetable object, or a list of datasets. schedule_interval is the older name.

start_date is the earliest interval the DAG will consider. It is not "the time of day the job runs." For a daily cron 0 6 * * *, Airflow creates a run when the interval has closed. A run with data interval [2026-03-01, 2026-03-02) becomes eligible at the end of that interval, aligned to the timetable. The practical effect you want is a morning run whose data interval is the business day you mean. Read the interval inside the task. Do not assume ds is "today."

In Airflow templates, ds is the logical date as YYYY-MM-DD, which for a regular schedule is the start of the data interval. data_interval_start and data_interval_end are the precise bounds. TechLearningPro treats data_interval_start converted to a date in UTC as business_date for this daily DAG. That rule is written in the DAG docstring so a future weekly DAG does not copy it blindly.

catchup=True means "create DagRuns for every interval from start_date to now that does not exist." catchup=False means "only the latest interval." Lesson 19's warning stands: a new DAG with start_date in 2020 and catchup=True will enqueue years of runs. TechLearningPro sets catchup=False on the daily DAG and backfills explicit ranges through the CLI when history is required. The default in Airflow is catchup=True. Set it explicitly in every DAG so a config change does not surprise you.

max_active_runs=1 is the lesson 19 cap for scheduled operation. max_active_tasks caps concurrent tasks inside the DAG.

Executors, briefly

LocalExecutor runs tasks as processes on one machine. Fine for a course environment and a small deployment. It shares fate with that machine.

CeleryExecutor sends tasks to a Celery broker and a pool of workers. Workers can scale. They must have the same DAG code and provider versions as the scheduler. A network filesystem or a git sync has to put dags/ on every worker.

KubernetesExecutor starts a pod per task and can give the fact load a different image or memory limit from the cleanse. Pod startup is slower. Burstiness is easier.

The DAG code should not care which executor is configured, except when you set a queue or a pool. Do not import Celery inside a task to "help." Pick the executor in deployment config.

Connections and variables

A Connection is a named set of host, schema, login, password, and extras, stored in the metadata database or fetched from a secrets backend. Tasks refer to conn_id="warehouse". The DAG file contains the id, not the password. A Variable is a named string or JSON blob for small config such as a bucket name. Variables are not a place for row data, and reading many Variables at import time hits the database on every parse. Read them inside the task if they must stay dynamic.

Production should use a secrets backend so passwords are not sitting in the Airflow database in reversible form. Lesson 21 shows the operational side. The authoring rule is already: conn_id in code, secret in the connection.

XCom

XCom is a cross-communication row: a task instance pushes a small value, a downstream task pulls it. The TaskFlow API pushes the return value of a @task automatically and injects it into downstream function arguments.

XCom lives in the metadata database by default. A return value of a million order lines will bloat that database and slow the scheduler. Return a path, a business date, and counts.

pythonCopy
return {
    "staged_table": "staging.order_line_winners",
    "business_date": "2026-03-01",
    "row_count": 18420,
    "quarantine_count": 17,
}

There is an extension point, BaseXCom, for storing the payload in object storage and keeping a pointer in the metadata database. Use it when you truly must pass a large object, and still prefer a table. XCom is not a bus. It is a convenience for small, named values.

XCom values are tied to the task instance. A backfill for another date has another XCom. Pulling "the latest" without a task instance id is how a March run reads January's path. TaskFlow wiring avoids that class of bug by passing the value along the edge.

Operators and the TaskFlow API

An operator is a template of work: run Python, run SQL, run a bash command, empty node. In Airflow 2, PythonOperator lives in airflow.operators.python. EmptyOperator replaced DummyOperator in 2.3. Bash is available and is the wrong place for warehouse logic, because a shell script hides exit codes and SQL.

The TaskFlow API (@dag, @task) is Python-native: each decorated function is a task, and using the return value sets the dependency and the XCom. Classic operators remain the right tool for SQL files via SQLExecuteQueryOperator, where the work is a statement rather than a Python function.

You can mix them. The daily DAG below uses TaskFlow for Python steps and a SQL operator for the reconciliation statement.

What the worker receives

The worker process loads the task and calls your function with a context. Important keys:

• data_interval_start, data_interval_end — pendulum datetimes
• ds — logical date string
• run_id — DagRun id
• ti — the task instance, if you must push an extra XCom

In a @task, call get_current_context() rather than expecting every parameter to be filled from context. Declare arguments you want Airflow to fill from upstream XComs.

provide_context is not a switch you must remember to set. Airflow 2 provides context to Python callables that accept it.

Idempotency at the DAG boundary

retries on a task will re-invoke the callable. The DAG does not wrap your SQL in a magic exactly-once box. Keep the merge. Also set retries in default_args so a new task inherits the policy from lesson 19 instead of an unexamined default.

Serialization limits

The DAG processor serializes the DAG to JSON in the database. Task callables must be importable: module-level functions, not lambdas and not closures that capture a database handle. Decorated @task functions in the DAG module are the normal pattern. A callable defined inside another function often fails serialization or points at the wrong object on the worker.

Default arguments must be serializable. A database connection object in default_args will fail. A conn_id string will not.

Architecture and Design Diagrams

mermaidCopy
flowchart TD
    files[DAG Python files] --> processor[DAG processor]
    processor --> meta[(Metadata database)]
    scheduler[Scheduler] --> meta
    scheduler --> executor[Executor]
    executor --> workers[Workers]
    workers --> meta
    workers --> warehouse[(Warehouse)]
    workers --> lake[Landing files]
    web[Web server] --> meta
    triggerer[Triggerer] --> meta

The warehouse is not inside Airflow. Workers open a connection by conn_id and run the merge. The metadata database holds the fact that the task succeeded, not the fact table.

mermaidCopy
flowchart TD
    landFile[land_reseller_file] --> cleanse[cleanse_orders]
    landFeed[land_learner_feed] --> changes[build_learner_changes]
    changes --> dimLearner[load_dim_learner]
    landCatalog[land_course_catalog] --> dimCourse[load_dim_course]
    cleanse --> fact[load_fact_order_line]
    dimLearner --> fact
    dimCourse --> fact
    fact --> reconcile[reconcile_revenue]
    reconcile --> publish[publish_freshness]

This is the lesson 19 graph as task ids. Airflow will not infer it from table names. You write the edges.

Technical Explanation

A minimal DAG before the full load

This DAG does no warehouse work. It shows the objects. It is valid Airflow 2.4+ TaskFlow, and the style matches 2.9.

pythonCopy
from __future__ import annotations

import pendulum
from airflow.decorators import dag, task

@dag(
    dag_id="tlp_skeleton",
    schedule="0 6 * * *",
    start_date=pendulum.datetime(2026, 3, 1, tz="UTC"),
    catchup=False,
    max_active_runs=1,
    tags=["techlearningpro"],
)
def tlp_skeleton():
    @task
    def show_window() -> dict:
        from airflow.operators.python import get_current_context

        ctx = get_current_context()
        start = ctx["data_interval_start"]
        end = ctx["data_interval_end"]
        return {
            "business_date": start.date().isoformat(),
            "interval_end": end.isoformat(),
            "run_id": ctx["run_id"],
        }

    show_window()

tlp_skeleton()

The last line instantiates the DAG. Forgetting to call the decorated function defines nothing. The processor imports the module, the call runs, and the DAG registers itself.

business_date is the UTC date of data_interval_start. For schedule="0 6 * * *", confirm in the grid that the interval is the one you think it is before anyone depends on it. Print it in this skeleton on a staging deployment. If the timetable produces an interval you did not expect, fix the timetable here, not inside the merge.

Dependencies and trigger rules

Calling a TaskFlow task with another task's output sets an edge and an XCom pull:

pythonCopy
summary = cleanse_orders()
load_fact_order_line(summary)

Multiple upstreams:

pythonCopy
cleaned = cleanse_orders()
learner = load_dim_learner()
course = load_dim_course()
load_fact_order_line(cleaned, learner, course)

The fact task will not start until all three have succeeded. That is trigger_rule="all_success", the default. You do not need to set it for this DAG. Other rules (all_done, one_failed, none_failed_min_one_success) exist. Using all_done on the fact load would start it after a failed dimension load. Do not.

Classic bitshift still works with operators:

pythonCopy
land >> cleanse >> [load_dim_learner, load_dim_course] >> fact >> reconcile

Lists mean fan-out and fan-in. Mixing styles is legal. Pick one style per DAG so reviewers can see the graph in one glance. The production example below stays in TaskFlow plus one SQL operator.

SQLExecuteQueryOperator

SQLExecuteQueryOperator is in airflow.providers.common.sql.operators.sql. It runs a statement on a conn_id. The SQL field is templated, so {{ data_interval_start }} renders for that task instance.

pythonCopy
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

reconcile = SQLExecuteQueryOperator(
    task_id="reconcile_revenue",
    conn_id="warehouse",
    sql="sql/reconcile_order_lines.sql",
)

The SQL file should fail the statement when the difference is nonzero, for example by raising an exception inside a DO block. A SELECT that returns a mismatch row does not fail the operator by itself. The operator fails when the database raises or the connection errors. Design the SQL to RAISE EXCEPTION when the diff is nonzero, so a green task means the sums matched.

Templating and typed intervals

Rendered templates are strings unless you set render_template_as_native_obj=True on the DAG. A Python task that needs real datetimes should use get_current_context() and the pendulum objects, which are already typed. Do not parse ds back from text if the context already has the interval.

Retries in default_args

pythonCopy
default_args = {
    "owner": "data-platform",
    "retries": 2,
    "retry_delay": pendulum.duration(minutes=5),
    "retry_exponential_backoff": True,
    "max_retry_delay": pendulum.duration(minutes=30),
}

Override retries on reconcile_revenue to 0. Lesson 21 goes deeper on backoff and callbacks. Setting them in the definition is part of authoring the DAG, so they appear here once.

Access control

access_control on the DAG can limit which roles see it. The web server enforces RBAC. A DAG that loads revenue should not be editable by every analyst login. Read access to the grid is useful. Write access to clear a task is a production change: clearing a task reruns it. Treat the Airflow UI permission to edit a DAG as production write access to the warehouse, because that is what a cleared fact task will do.

Code layout

textCopy
dags/
  tlp_orders_daily.py      # graph only, thin wrappers
  sql/
    reconcile_order_lines.sql
include/
  tlp_transforms.py        # parse_line, pure functions, no Airflow imports
tests/
  test_transforms.py

Pure functions stay importable without an Airflow install so lesson 23 can test them quickly. Import a heavy client inside the task if the import is slow. An expensive import at module top slows every parse.

Practical Implementation or Design Walkthrough

The following module is a complete DAG definition for the daily load. The transform bodies are the lesson 18 functions, called by name. SQL files are referenced, not inlined, so the DAG stays a graph. This targets Airflow 2.9 with apache-airflow-providers-common-sql installed. On 2.4 through 2.8 the same schedule, @dag, and @task forms apply.

pythonCopy
"""Daily TechLearningPro warehouse load.

business_date is the UTC date of data_interval_start.
Dimensions run before facts. Reconciliation raises on a nonzero diff.
"""

from __future__ import annotations

import pendulum
from airflow.decorators import dag, task
from airflow.operators.python import get_current_context
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

def _window() -> dict:
    ctx = get_current_context()
    start = ctx["data_interval_start"]
    return {
        "business_date": start.date().isoformat(),
        "interval_start": start.isoformat(),
        "interval_end": ctx["data_interval_end"].isoformat(),
        "run_id": ctx["run_id"],
    }

@dag(
    dag_id="tlp_orders_daily",
    schedule="0 6 * * *",
    start_date=pendulum.datetime(2026, 3, 1, tz="UTC"),
    catchup=False,
    max_active_runs=1,
    default_args={
        "owner": "data-platform",
        "retries": 2,
        "retry_delay": pendulum.duration(minutes=5),
        "retry_exponential_backoff": True,
        "max_retry_delay": pendulum.duration(minutes=30),
    },
    tags=["techlearningpro", "warehouse"],
    doc_md=__doc__,
)
def tlp_orders_daily():
    @task
    def land_reseller_file() -> dict:
        from tlp.ingest import land_reseller

        return land_reseller(**_window())

    @task
    def land_learner_feed() -> dict:
        from tlp.ingest import land_learner_feed as land

        return land(**_window())

    @task
    def land_course_catalog() -> dict:
        from tlp.ingest import land_course_catalog as land

        return land(**_window())

    @task(retries=1)
    def cleanse_orders(landed: dict) -> dict:
        from tlp.cleanse import cleanse_reseller_file

        return cleanse_reseller_file(landed, **_window())

    @task
    def build_learner_changes(landed: dict) -> dict:
        from tlp.cleanse import learner_changes_from_feed

        return learner_changes_from_feed(landed, **_window())

    @task
    def load_dim_learner(changes: dict) -> dict:
        from tlp.warehouse import apply_learner_scd

        return apply_learner_scd(changes, **_window())

    @task
    def load_dim_course(landed: dict) -> dict:
        from tlp.warehouse import apply_course_scd

        return apply_course_scd(landed, **_window())

    @task
    def load_fact_order_line(cleaned: dict, learner: dict, course: dict) -> dict:
        from tlp.warehouse import merge_fact_order_line

        return merge_fact_order_line(cleaned, learner, course, **_window())

    @task(retries=0)
    def publish_freshness(recon_ok: list) -> dict:
        from tlp.warehouse import upsert_freshness

        # recon_ok is the XCom from the SQL operator. The statement raised if the diff was nonzero.
        return upsert_freshness(recon_ok, **_window())

    reseller = land_reseller_file()
    feed = land_learner_feed()
    catalog = land_course_catalog()
    cleaned = cleanse_orders(reseller)
    changes = build_learner_changes(feed)
    learner = load_dim_learner(changes)
    course = load_dim_course(catalog)
    fact = load_fact_order_line(cleaned, learner, course)

    reconcile = SQLExecuteQueryOperator(
        task_id="reconcile_revenue",
        conn_id="warehouse",
        sql="sql/reconcile_order_lines.sql",
        retries=0,
    )
    fact >> reconcile
    publish_freshness(reconcile.output)

tlp_orders_daily()

SQLExecuteQueryOperator.output is the XCom of that task, so publish_freshness waits for reconciliation. retries=0 on that operator matches the lesson 19 rule. The SQL file uses templates:

sqlCopy
-- sql/reconcile_order_lines.sql
-- Fails the task when warehouse cents disagree with staged winners.
DO $
DECLARE
    diff bigint;
BEGIN
    SELECT COUNT(*) INTO diff FROM (
        SELECT 1
        FROM analytics.fact_order_line f
        FULL JOIN staging.order_line_winners s
          ON s.order_line_id = f.order_line_id
        WHERE COALESCE(f.extended_amount_cents, 0) <> COALESCE(s.extended_amount_cents, 0)
          AND COALESCE(f.business_date, s.business_date) = DATE '{{ data_interval_start }}'
    ) mismatches;
    IF diff <> 0 THEN
        RAISE EXCEPTION 'recon_diff_rows=%', diff;
    END IF;
END $;

DATE '{{ data_interval_start }}' depends on the rendered value casting cleanly. If rendering includes a time component, cast in SQL with an explicit timestamp cast you have tried on a staging database. Verify the rendered SQL in the task instance details before production. Templating bugs are visible there: you can read the statement Airflow actually sent.

Setup. Airflow 2.9, PostgreSQL metadata database, LocalExecutor for a single-node course lab. Create the connection warehouse in the UI or with airflow connections add, taking the URI from the environment, not from the DAG. Put dags/ on the scheduler's DAG folder. Confirm airflow dags list-import-errors is empty. Unpause the DAG. Trigger one run manually so you do not wait until 06:00. Open the grid, confirm one DagRun, and confirm each task instance shows the same data_interval_start.

What success looks like. publish_freshness is green. ops.dataset_freshness has a row for that business date with a zero reconciliation difference. A second clear of load_fact_order_line reruns the merge and leaves one row per order_line_id.

What this DAG does not do yet. It does not wait with a sensor if the file is late, it does not use pools, it does not emit dataset events, and it does not define a backfill procedure. Those are lesson 21, on this same dag_id.

Enterprise Real-World Scenarios

One production Airflow, many domains

TechLearningPro's platform team runs Airflow for finance loads, product analytics, and a nightly catalog sync. Each domain has a DAG owner in default_args and a tag. The metadata database is sized as a small, fast PostgreSQL instance with backups, because losing it loses run history and the ability to schedule. Workers use CeleryExecutor across two queues: warehouse and default. The fact DAG's tasks go to warehouse so a burst of experimental DAGs cannot occupy every slot. Code is deployed by CI, which lesson 23 will gate. Analysts can view the grid. Only the platform on-call role can clear a task in production. Clearing load_fact_order_line is a warehouse write, and the permission model treats it that way.

The first outage in this deployment was a DAG file that opened a warehouse connection at import. The warehouse stalled, the processor could not import the file, and the scheduler logs filled with import timeouts. Unrelated DAGs in other files kept running. The broken file did not. The fix was to move the query into the task. The review rule that followed: no hooks at module top. A CI check imports the DAG folder with example DAGs disabled and fails on any import error. That check is necessary and still does not prove the absence of a lazy connection. Code review has to see the import.

A logical-date bug that mis-labeled a Monday

A change replaced _window() with business_date = date.today() - timedelta(days=1) to "make the SQL easier." On Tuesday the run was late and started after midnight UTC. Tasks in the same DagRun computed different dates depending on when each process started. The merge key saved revenue from duplication. The freshness row pointed at the wrong date, and the late file's real interval was never requested because the task ignored it. Restoring data_interval_start and backfilling the missing date fixed the data. The review rule: if a task body uses date.today or datetime.now as a business date, it does not ship. The database clock may stamp loaded_at. It may not choose the business date.

Handing the same task boundary to Spark

When reseller volume outgrows the worker's Python process, cleanse_orders will submit a Spark job instead of calling cleanse_reseller_file in-process. The DAG id, the task id, the interval, and the downstream edge to load_fact_order_line stay. The Spark job writes staging.order_line_winners for that business date and returns the same small dict. Module 3's Big Data Fundamentals and the Spark lessons own the cluster. This DAG owns the decision that reconcile still runs after the fact load and still fails the task on a nonzero diff. Orchestration does not move into the Spark driver as a second scheduler. One control plane is enough.

Best Practices

• Keep DAG files free of I/O at import time. Task ids must be stable across parses.
• Set schedule, start_date, and catchup explicitly. Prefer catchup=False plus deliberate backfills.
• Derive business_date from data_interval_start or data_interval_end under a written rule.
• Use TaskFlow return values for small summaries. Store rows in tables.
• Put SQL that can RAISE on a failed reconcile in a file the operator runs, so a green task means the check passed.
• Pin Airflow and provider versions. Ship the same code to scheduler and workers.
• Reference conn_id. Keep passwords out of Git.
• Set owner, retries, and max_active_runs in the DAG instead of inheriting an unknown default.
• Give the pipeline one publish task downstream of reconcile.
• Separate pure transforms from the DAG module so they test without a scheduler.
• Read the rendered template and the attempt log before changing a query.
• Restrict UI clear and edit permissions. They are production writes.
• When a task grows into Spark, keep the task boundary and the XCom contract. Change the body.

Common Mistakes and Anti-Patterns

Warehouse calls at import. The processor depends on the warehouse, parses slow down, and a source outage stops scheduling that file. Move the call into the task.

catchup=True with an old start_date. Deploy day creates hundreds of DagRuns. Set catchup=False unless you have capacity and a reason.

Using datetime.now() as the business date. Late runs and backfills drift. Use the data interval.

Returning a DataFrame from @task. The metadata database fills with XCom blobs and the scheduler slows. Return counts and a table name.

A bash script that swallows SQL errors. The task is green and the table is empty. Use a SQL operator or a Python function that raises.

trigger_rule="all_done" on a load. The fact load runs after a failed parent. Keep the default all_success.

Two DAG definitions with the same dag_id in different files. The processor loads one of them and the grid does not match the file you edited. dag_id is unique on the deployment.

Editing a production DAG's task ids during an incident. Running task instances point at ids that no longer exist and show as removed. Add a new id, migrate, then delete.

Sharing one metadata database between dev and prod. A test DAG with conn_id="warehouse" can point at production if connections were copied. Separate deployments.

Storing the database password in a Variable. Variables show up in the UI and can land in logs when templated. Use a Connection or a secrets backend.

Lambdas as operators. Serialization fails on the worker or points at the wrong function. Use module-level functions or @task.

Troubleshooting and Design Review

Design review for a new DAG:

1. Does import do any I/O?
2. Is catchup explicit, and is start_date the first interval you will own?
3. Where does business_date come from?
4. Do edges match the lesson 19 graph, with reconcile before publish?
5. What is returned on XCom, and is it small?
6. Which tasks have retries=0 because failure is a data outcome?
7. Which conn_id values are used, and are the secrets outside the repo?
8. Can the pure functions be imported without starting Airflow?

Interview Questions and Answers

Q1. What processes make up an Airflow 2.9 deployment, and which one runs your SQL?

The DAG processor imports files and writes serialized DAGs. The scheduler creates DagRuns and queues task instances. The executor starts work. The worker process runs your SQL or Python. The triggerer handles deferrable waits. The web server is the UI and API. The metadata database stores state. The worker runs the SQL against the warehouse. The scheduler does not. Stopping the web server does not stop scheduling.

Q2. How is a DAG different from a DagRun and a task instance?

The DAG is the definition, including tlp_orders_daily and its edges. A DagRun is one interval's execution of that definition. A task instance is one task inside that run, with attempts and state. Retries do not create a new DagRun. They add attempts to the same task instance. Logs and XComs are per task instance, so a March run does not read a January XCom unless you point at that instance.

Q3. Why must DAG import stay free of warehouse queries?

The processor imports the file on a cadence, on the scheduling side, not as part of the data interval. A query at import runs far more often than the daily load, couples scheduling to warehouse health, and can change task structure between parses. Run the query inside a task, where it has an interval, a log, and retries.

Q4. What do schedule, start_date, and catchup mean together?

schedule is the timetable, such as 0 6 * * *. start_date is the earliest interval Airflow may create. catchup=True fills every missed interval from that start until now. catchup=False schedules the latest interval only. A new production DAG at TechLearningPro sets catchup=False and an honest start_date, then backfills a chosen range on purpose. Leaving catch-up on with a start date years in the past is how a deploy creates a queue you did not plan to run.

Q5. What is data_interval_start used for in tlp_orders_daily?

It is the start of the interval that DagRun is responsible for. The DAG defines business_date as that instant's UTC date and passes it into land, cleanse, and merge. Tasks do not call date.today(). ds is the logical date string and often matches that start, and the explicit interval is what you log and test. A late wall clock does not change the interval.

Q6. What belongs in XCom, and what does not?

A path, a table name, a business date, and row counts belong there. A DataFrame, a file body, and a password do not. XCom is stored in the metadata database by default and is tied to the producing task instance. TaskFlow passes return values along edges so the downstream task pulls the right instance. Large payloads belong in staging tables. A custom BaseXCom backend can move blobs to object storage if you have a measured reason.

Q7. How do you express "fact load waits for both dimensions and the cleanse"?

Pass all three upstream return values into load_fact_order_line, or use bitshift fan-in with classic operators. The default trigger rule all_success waits until every parent has succeeded. all_done would start the load after a failed dimension task, which publishes facts against a broken dependency. The edge is the lesson 19 dependency, written in code.

Q8. Why did SQLExecuteQueryOperator get retries=0 on reconcile?

A nonzero reconciliation difference is a business failure. Retrying it does not change the cents unless something else is changing the tables underneath you. The SQL raises an exception so the task cannot turn green on a mismatch that was merely returned as a row. Land and load tasks keep retries because their writes are idempotent and their typical faults are locks and timeouts.

Q9. LocalExecutor, CeleryExecutor, and KubernetesExecutor: what changes for the DAG author?

Where the task process runs, and how you scale and isolate it. The DAG still declares tasks and edges. Workers must have the DAG code and providers. KubernetesExecutor can give a task its own pod resources. CeleryExecutor uses queues. LocalExecutor shares the scheduler machine and is appropriate for a lab and some small deployments. The author does not open a broker from inside the parse path to send the task.

Q10. Where do warehouse passwords live?

In an Airflow Connection referenced by conn_id, ideally backed by a secrets backend in production. Not in the DAG file, not in a templated command, and not in an XCom. The task log should show the connection id. UI permissions to edit connections are production-secret permissions. Dev and prod deployments use different metadata databases so a copied DAG cannot reach production credentials by accident.

Q11. A teammate clears load_fact_order_line in the UI. What happens?

Airflow schedules a new attempt of that task instance for the same data interval. Downstream tasks may be cleared with it if the operator chose to clear downstream. The merge runs again. Because the merge is idempotent, the fact table stays at one row per order_line_id. If the task were not idempotent, the clear would double revenue. UI permission to clear is warehouse write access, and the data-plane merge is what keeps the click safe.

Q12. How does this DAG stay valid when cleanse moves to Spark?

cleanse_orders keeps its task id, its upstream land task, and its downstream fact load. The body submits Spark and waits for the job that writes the same staging table for the same business_date, then returns the same summary dict. Reconcile still raises on a diff. Airflow remains the only scheduler. The pure functions and the SQL contract are what Module 3 must reproduce, which is why they do not live only inside the DAG file.

Hands-On Exercises

Exercise 1. Skeleton and interval

Problem. Run tlp_skeleton on a local Airflow 2.9 with LocalExecutor.

Tasks.

1. Install a pinned apache-airflow 2.9 release with the official constraint file for your Python version.
2. Register the DAG. Clear import errors.
3. Trigger a run and record data_interval_start, data_interval_end, and run_id from the task log or XCom.
4. State the business_date your rule produces.
5. Explain what catchup=False prevented, given your start_date.

Acceptance. The DAG appears without import errors, and the business date comes from the interval, not from the wall clock of the trigger click.

Exercise 2. Wire the real graph

Problem. Implement tlp_orders_daily against a local PostgreSQL warehouse with one order line and one learner change.

Tasks.

1. Create connection warehouse.
2. Implement thin tasks that call the lesson 18 merge and SCD change insert, or SQL equivalents.
3. Make reconcile_order_lines.sql raise when the fact cents differ from staging.
4. Run the DAG twice for the same logical date by clearing load_fact_order_line only. Show one fact row.
5. Fail the dimension task on purpose and show that the fact task does not run.

Rubric.

Lesson Summary

Airflow 2.9 turns the lesson 19 graph into a DAG definition, DagRuns for each data interval, and task instances that workers execute. The processor must be able to import the file cheaply. The scheduler queues work. The worker runs the merge. XCom carries counts and names. The warehouse carries rows. data_interval_start is the business clock. catchup is a decision you set on purpose. Reconciliation is a task that raises, and publish sits downstream of it.

The DAG in this lesson is complete enough to run the daily load and thin enough to test. The next lesson operates it: retries under load, backfills, dataset scheduling, pools, sensors, and the metrics that tell you the scheduler is healthy. The SQL does not move into those features. They are how the same DAG stays reliable on a bad morning.

Further Reading

• Apache Airflow 2.9, DAG runs — logical date versus data_interval_start and data_interval_end.
• Apache Airflow 2.9, TaskFlow — @dag and @task, and automatic XCom from return values.
• Apache Airflow 2.9, XComs — what the metadata database stores when you return a value.
• Apache Airflow providers, SQLExecuteQueryOperator in apache-airflow-providers-common-sql — the operator used for reconciliation. Pin the provider you install.
• Apache Airflow 2.9, Connections — conn_id as the credential handle in DAG source.
• TechLearningPro, Data Pipeline Orchestration Fundamentals — the graph this DAG implements.
• Next: Advanced Airflow Operations.