DAG Anatomy: Intermediate

In Airflow 3 one component owns your DAG files: the dag processor. It walks the dags folder, imports each .py file as a Python module, collects every DAG object that import produced, and writes a serialized copy of each DAG into the metadata database. The scheduler and the web UI read that serialized copy and never open the file. When the processor reaches the end of the folder it returns to the start and imports everything again, and it keeps doing that for as long as it runs. Everything in this lesson follows from that loop: what it costs, why a file can fall out of it, and why a test on your laptop can tell you what the processor will do.

The [dag_processor] section of the config has 3 settings that set the pace, and their defaults matter because most deployments keep them. A file is re-imported at most every 30 seconds, 2 processes import files side by side, and a single import that runs past 50 seconds is killed. On a small folder those numbers mean the top of your file runs around 2,880 times a day, and each of those runs has to finish inside 50 seconds or the DAG never reaches the database.

Every statement at module level runs on every import. The body of a @task function stays unexecuted, because constructing the task objects already gives the processor what it needs: which operator each task id uses, and what it depends on. So a file that opens a warehouse connection at module level, intending the task to reuse it, pays that handshake once per import, and the task never sees the connection anyway, because the task runs in a different process on a different machine.

An exception at module level stops the import, and the processor records it against the file path in the import error table. The UI shows an import error notice on the home page, and airflow dags list-import-errors prints the file path with the full traceback. Missing modules and mistyped names land here, as does a KeyError on an environment variable the processor's host lacks. A file that errors produces no DAG, so a new DAG from a broken file is absent from the list rather than shown in red.

The processor can skip a file without ever opening it, and it records no error when it does. Before importing, it scans the file's text for the strings airflow and dag, case-insensitive, and skips any file lacking either one. A file that gets its DAG by calling a factory function from another module can fail this test and sit there in silence. A .airflowignore file in the folder or any parent folder also holds glob patterns, and a file matching one is skipped, along with everything under a matching directory. And the folder the processor scans is whatever airflow config get-value core dags_folder prints, so a file 1 directory outside it does not exist to Airflow.

The processor collects DAG objects from the module's global namespace after the import. A @dag decorated function that is defined and never called creates no DAG, because the decorator turns the function into a factory, and the DAG exists only when the factory runs. A with DAG(...) block inside a function the module never calls has the same effect. In both cases the import succeeds, there is no error to list, and the file parses in milliseconds, so this cause hides behind a clean report.

A dag_id is unique across the whole database. If 2 files define the same id, both import cleanly. The DAG then shows under whichever file the processor parsed last, with that file's tasks. The other file's tasks never appear, and the DAG's recorded file path can change from one sweep to the next. Copying a DAG file to start a new one and forgetting to change the id produces this exactly, and airflow dags list shows the id once, with a file path you did not expect.

An import that passes 50 seconds is killed and the DAG is never written. The dag processor's log records it with a line like Processor for dags/bucket_inventory.py with PID 4182 started 51 ago killing it., which names the file path and its age in seconds, past the 50-second limit. Causes 2 and 3 leave no line in that log at all, which is itself a clue. The usual source of a timeout is the module-level work from the previous section: a slow query, a listing of a large bucket, an import chain that pulls in a numerical library. The fix is the same as before, moving the work into a task, and the check is airflow dags report, which prints the file's parse duration next to its name.

A DAG that appears with its toggle off has arrived and is paused. New DAGs start paused by default, which is a different situation from a DAG that never arrived.

After fixing the file, you can wait for the next sweep or force one. airflow dags reserialize reparses every file in the folder and rewrites the serialized DAGs, which is the quickest way to confirm a fix on a deployment where sweeps run slow. If the DAG still fails to appear after a forced reserialize, the cause is one of the 5 above, and the order of checks will find it.

The apache-airflow package installs the scheduler and the dag processor, plus the API server behind the UI and the command line. It also installs the Task SDK: you import DAG from airflow.sdk, and dag, task and TaskGroup come from the same place. Operators live elsewhere, in provider packages named apache-airflow-providers-<name>, each released on its own schedule and installed with pip. Even BashOperator and PythonOperator live in a provider, apache-airflow-providers-standard. Core depends on that one, so it is always present, and its import paths show the pattern every other provider follows.

Each path's second segment names the package, so standard comes from apache-airflow-providers-standard. The same reading gives apache-airflow-providers-http for http and apache-airflow-providers-common-sql for common.sql. To reach a system the folder cannot reach today, you install its provider, and airflow providers list prints every installed provider with its version. The provider for a particular warehouse or object store carries the operators for that system plus the plumbing they use to connect, and most SQL providers build on common-sql, so installing one pulls the other in.

The extras on the core package are a shortcut for the same install: pip install "apache-airflow[http]" brings in core and the http provider together, and the extra's name matches the provider's. Either route ends with the same package in the environment, so pick one route and use it everywhere.

A @task function runs any Python you write with the Task SDK alone, and for a transformation that is your own code it is the shortest route. A provider operator is the better choice when the work is a conversation with an outside system: it wraps that system's client with authentication and error handling already written, and it writes a task log that reads well, so a 20-line @task that hand-rolls an HTTP call usually loses to a 4-argument HttpOperator. Reading an unfamiliar operator starts with its constructor arguments, which are the knobs you set, and its execute method, which is what the worker runs. For BashOperator, execute runs the command and raises if the exit code is non-zero, so the task fails when the script fails.

python dags/orders_health.py imports the module the way the processor does. Zero output and exit code 0 mean every module-level statement ran without raising. That alone does not prove a DAG exists: an uncalled @dag function imports cleanly and builds nothing, as cause 3 showed. airflow dags list naming orders_health, or the depth 3 run below, confirms the DAG was built. A traceback is the same one the processor would record, shown now rather than after the next sweep. Task bodies stay unexecuted during an import, so a passing import says nothing about whether check_orders works, and a slow import on your laptop predicts a slow parse on the deployment.

airflow tasks test orders_health check_orders 2026-11-03 runs that single task in your terminal. It ignores the task's dependencies, so upstream tasks do not run first. It prints the task log to the screen. And it records no state to the metadata database, so nothing appears in any grid afterwards. That makes it the command to iterate on one function: edit, run, read the log, repeat, on a 5-second loop. Its blind spot follows from the first point: anything the task expects to receive from an upstream task is absent, because upstream never ran.

airflow dags test orders_health 2026-11-03 builds the DAG, creates 1 run stamped with that date, and executes every task in dependency order inside the one process. No scheduler or worker takes part. Task logs stream to the terminal. This command does write to the metadata database: the run and its task states are saved, and the grid shows the run afterwards. A task that fails stops its downstream tasks from running, and the command exits non-zero, so a shell script or a CI job can act on the exit code.

The date 2026-11-03 is the date the run is stamped with. What that date means for the data a run reads belongs to the scheduling lessons. When testing, reuse one date so that 2 runs can be compared.

--mark-success-pattern takes a regular expression, and any task whose id matches is marked success without running, so airflow dags test orders_health 2026-11-03 --mark-success-pattern "load_.*" skips every load task while you check what comes after them. A 20-minute wait becomes 20 seconds, and the tasks you care about still run in their real order.

The processor imports the file as a module, so the __main__ block never runs there and costs nothing. Once the file carries that block, running it directly becomes depth 3 with your debugger attached, and python -c "import orders_health" from inside the dags folder gives you depth 1 again.

A DAG with 30 tasks draws a graph nobody reads. If 12 of those tasks pull from source systems, 12 clean what arrived and 6 publish, the graph has 3 shapes in it, and a TaskGroup lets you draw those 3 shapes as 3 nodes. The group is a naming and drawing device: it changes task ids and how the UI folds the graph, and leaves what each task runs untouched.

Each task inside a group got the group's name prefixed to its id, so the task you wrote as pull_orders is stored and shown as extract.pull_orders, and that full id is what every command takes, so testing it is airflow tasks test orders_daily extract.pull_orders 2026-11-03. extract >> clean also wired the groups: every task in extract with nothing downstream inside its group now points at every task in clean with nothing upstream inside its group. Here that is 3 tasks pointing at 1, since cast_types already has dedupe upstream, so the single arrow created 3 edges.

@task_group wraps a function the way @dag wraps a DAG: the group exists when the function is called, and the call returns an object you can wire with >>. The decorator takes the same arguments as TaskGroup. tooltip shows when the pointer rests on the collapsed node. prefix_group_id=False keeps the inner ids bare, at the cost of making you keep them unique across the whole DAG yourself.

Arrows need not stop at group borders. clean >> notify, with notify a plain task, hangs that task off every leaf of the group, and an inner task can be wired on its own: keep the return value of pull_orders() in a variable and write orders >> some_task to join 2 tasks in different groups without touching the rest of either group.

Putting existing tasks into a group renames them. pull_orders becomes extract.pull_orders, and the metadata database has task history under pull_orders and nothing under the new id. The UI shows the old id as a removed task and the new one with an empty history, and any script that calls airflow tasks test with the old id now fails to find it. For a DAG with a long history worth keeping, prefix_group_id=False on the group gives you the folded graph with the ids unchanged. For a new DAG, let the prefix happen and keep group names short, since every task id carries them from then on.