DAG Anatomy: Advanced

The one new call is load.expand(path=list_files()) in place of load(path=...). At parse time the processor sees 3 tasks and the graph draws 3 boxes, one each for list_files, load and report. Nothing has fanned out yet, because the parser never calls list_files. When a run is triggered and list_files finishes, the scheduler reads its return value. It finds 3 strings and creates 3 task instances of load, numbered 0 to 2 by map_index. Each copy receives one string as its path. The grid keeps load on a single row and labels the row with the count; opening it lists the copies with their indexes and states.

report is an ordinary task that takes names. Because its argument is the output of a mapped task, it waits for all 3 copies to finish, then receives their return values as a sequence in index order: ['acme.csv', 'globex.csv', 'initech.csv']. If 1 copy fails, report never runs, the same as it would behind a single failed task. A downstream task wired with >> and no argument from load waits for every copy in the same way, because the mapped task counts as done only when its last copy is.

expand takes the arguments that vary per copy. Everything shared across copies goes in partial, which is also where a classic operator receives its task_id, since the operator is never constructed directly. Passing 2 lists to expand gives the cartesian product: 3 paths times 2 formats is 6 copies. When the values must stay paired, expand_kwargs takes a list of dicts, 1 dict per copy, each holding that copy's complete keyword set. A name given to partial cannot also be given to expand; the file fails to parse if it is.

Inside a mapped task, get_current_context()["ti"].map_index returns the copy's index, which lets a copy write to a file or a partition named after itself without colliding with its siblings. An unmapped task reads -1 there. The context holds no total, so a copy that must know how many siblings it has is handed the number: the upstream task returns 1 dict per file carrying both the path and len() of the list, and expand_kwargs passes each dict to its own copy. A task group can be mapped too: decorate it with @task_group, give it 1 argument, and call .expand() on the group; every task inside is copied once per value, and the grid shows the group with the count on its row, so a 3-step pipeline per file stays readable as 1 group with a number beside it.

Mapping fits when the copies share a run and an owner and can sit on one row in the grid, which describes the partner feeds. The warehouse team's loads do not fit. The team extracts and then loads 14 source tables the same way, but the payments team owns 2 of them, the search team owns 5, each table has its own history to inspect, and when a source breaks the owning team wants to pause that 1 load and leave the other 13 running. That calls for 14 DAGs, each with its own dag_id, and nobody wants to maintain 14 near-identical files. The answer is 1 Python file that reads a config and produces a DAG object per entry.

The processor imports this file like any other. The loop runs, build returns a DAG for each entry, and globals() places each under its own name at the top level of the module, which is where the processor collects DAG objects from. airflow dags list then shows load_orders and load_listings as 2 rows with the same file location. Each has its own pause switch, its own runs and its own grid; the shared file is invisible to anyone reading the UI. The entry dict is captured by the inner functions, so extract in load_orders builds its SQL from orders_db and nothing else, and adding a 3rd entry to the YAML produces a 3rd DAG on the next parse without touching the Python.

An error has the same blast radius. One entry with a missing source key raises KeyError inside build. That fails the import, and the import errors page lists the file once. The 14 DAGs already registered stay in the list on their last good definitions and stop updating there: no fix to any of them lands, an entry added in the same edit never appears, and a worker that loads the file to run one of their tasks hits the same error. Validating the config before the loop doesn't unfreeze the other 13, but it turns a KeyError on a line inside build into a message that names the broken entry, which is the difference between a 1-minute fix and a 20-minute hunt.

The subtler cost is identity. A DAG's history lives under its dag_id. Change the naming scheme from load_{name} to warehouse_{name} and the processor registers 14 new DAGs with empty histories, while the 14 old ones are marked as no longer present in any file and keep their runs under the old names. Treat the id formula as a contract, and keep the config file in the same repository as the DAG file so that a change to one is reviewed beside the other.

Since Airflow 3, the DAG processor doesn't watch a bare folder. It watches bundles, each a named source of DAG files with a class that knows how to fetch them. A fresh install has 1 bundle, dags-folder, of class LocalDagBundle, pointing at the dags folder on disk, which is why every earlier lesson worked without mentioning one. The list lives in the [dag_processor] section of the configuration as dag_bundle_config_list, and airflow dags list prints a bundle_name column so you can see which bundle each DAG came from.

A GitDagBundle, shipped in the git provider, tracks a branch of a repository. Its git_conn_id names the Airflow connection that holds the repository's URL and credentials, and without it the bundle has nothing to clone. Its version is the commit hash it last fetched, and it checks for a newer commit every refresh_interval seconds, 60 here. A LocalDagBundle has no version, because a folder on disk holds no record of which copy of a file it contained at 02:00. That one difference decides everything else in this section.

Every time the processor serializes a DAG and finds the result differs from the stored one, it records a new DAG version with a rising version number, and the same happens when the bundle's version changes under an unchanged file. The UI's Code tab lets you pick any version and read the file as it was then. Each run records the version it was created against, and each task instance records the version it executed under, which is what lets the grid show a run and say what code it was meant to execute. Any change to the tasks or the arrows between them, a renamed task id included, changes the serialized form; editing the body of a function inside a @task does not, because the serialized DAG holds the function's name and not its code.

A run of load_orders, whose 3 tasks are now extract, load and verify, starts at 02:00 against version 3. At 02:10 extract has finished and load is running when a deploy lands version 4, which fixes a bug in verify's body and adds an arrow from extract straight to verify. The question is which verify runs at 02:12, and the answer depends on the bundle.

With the local bundle, the worker opens whatever file is on disk when verify starts, so it runs the fixed body even though the run was created against version 3. The run has mixed code: extract and load from version 3, verify from version 4, and the task instance detail shows the version each one executed under. With the Git bundle, the run recorded commit c41, the worker fetches that commit before starting verify, and the run finishes on version 3 code throughout, bug included. The fix reaches the 02:30 run, which is created against version 4. Neither behavior is wrong; one gives you the fix 18 minutes sooner, the other gives you a run whose code you can name with 1 hash.

Airflow 3 draws a line between the parts that own the metadata database and the parts that run your code. The scheduler and the DAG processor hold database credentials, and so does the API server; they sit on one side. Workers sit on the other with none. A worker's only route to Airflow's state is HTTP to the API server, through a set of endpoints called the Execution API, and every request carries a token issued for 1 task instance and 1 try. The Task SDK, the airflow.sdk package, is the code that lives on the worker side: the names a DAG file imports, plus the machinery that runs a task and talks to the API on its behalf.

When a worker takes a task, it starts a supervisor process, and the supervisor starts the task itself in a child process. The supervisor asks the API server for the task's details, sends the heartbeat that tells the scheduler the task is alive, and relays any request the task makes. Each line the task prints passes to the supervisor, which writes it to the task's log file on the worker and, when the deployment has a remote log store, uploads that file there as the try ends; the UI's log view asks the API server, which fetches the lines from the worker or the remote store. The task sees a context dict whose entries are small objects built from the API's reply rather than database rows. ti.task_id and ti.map_index are attributes read from that reply; there's no session behind ti, and a call that assumes one fails.

The next file is what a task written for an older deployment often looks like. It opens a session and counts runs. On Airflow 3 the call to create_session raises RuntimeError with a message that direct database access from a task is not allowed, before any query is sent, because the worker has no connection string to send it on.

Losing the session does not leave a task blind to the rest of its run. The Execution API answers a short list of read requests, and the supervisor relays them: the value an upstream task returned, the number of task instances that match a filter through ti.get_ti_count, and their states through ti.get_task_states. The API server runs each lookup against the database and sends back only the result. A task can ask these questions; it can't write its own query, and it cannot change the state of any task instance but its own.

A worker runs whatever Python the DAG authors wrote, and a process that runs arbitrary code should not hold the credentials to a database that stores every DAG's state; the token scoped to 1 task instance is the smallest grant that still lets the task report back. airflow.sdk is also released as its own package, apache-airflow-task-sdk, with its own version number, so an operator written against it keeps working across core upgrades, and the database schema can change without any task noticing. And because a worker needs only HTTP to the API server, it can run on a host that cannot reach the database at all.

A @task function covers most needs: write Python, return a value, done. A custom operator is worth writing when the same body with the same 2 or 3 parameters is being pasted into DAG after DAG, when the task should look and behave like a built-in in the graph with its own color and rendered fields, or when stopping it from outside must clean something up. Write the function first, and promote it to an operator the 3rd time you paste it. The example below counts the rows in a CSV file and fails the task when the count is below a floor, a check 14 feed loads all need.

The class extends BaseOperator from airflow.sdk, which is what makes it a task the scheduler can plan and a worker can run. __init__ takes its own arguments as keywords and stores them on self. Every other keyword, task_id included, goes to super().__init__. It opens no file and reaches no service, because __init__ runs in the DAG processor on every parse cycle, 2 times a minute for as long as the file exists, and again on the worker each time it loads the file to run a task instance, once per try. A lookup placed there is paid by both processes on every one of those occasions. execute is the body: it runs once per try on a worker, receives the same context dict a @task function gets, and whatever it returns flows downstream the way a @task return value does. Raising any exception fails the try.

A DAG file calls the operator with CsvRowCountOperator(task_id="check_acme", path="feeds/acme.csv", min_rows=100), just as it would a built-in. Mapping it works the same way too. The file below checks every file list_files returns, which ties this section to the first one: partial carries task_id and min_rows, expand carries the paths, and the grid shows 1 row named check with the count of copies beside it.

Where the module lives matters because 2 different processes import it. The DAG processor imports it to build the task; the worker imports it to run execute. A module inside the bundle, dags/operators/csv_ops.py with an __init__.py beside it, reaches both, because both fetch the bundle and the bundle's root is on the import path. A package installed with pip on every host reaches both as well and is the right home once 2 bundles share the operator. A module that only the processor can see produces a DAG that parses and then fails on the worker with an import error at the top of every try.