DAG Anatomy: Beginner

A marketplace team needs a report of yesterday's bookings and reviews on a dashboard by 09:00. The first version is 3 scripts on a server's timer: export_bookings.py at 01:00, export_reviews.py at 01:15, and build_report.py at 02:00. The 02:00 slot is a guess that the 2 exports take under 45 minutes. One night in December the bookings export runs for 70 minutes because a holiday weekend tripled the rows. build_report.py starts at 02:00 on schedule, reads a half-written bookings file, and finishes with no error. The dashboard shows a 40% drop in bookings at 09:00, and the on-call engineer spends the morning finding out why.

The rest of this lesson uses 3 words. A DAG is one described workflow: the tasks and the order between them, defined in one Python file. A task is one unit of work in it, such as running a shell command. A run is one execution of the whole DAG for one scheduled time, so the DAG for the nightly report has one run per night.

How a task opens a connection to a database or an API, and how it hands a value to the task after it, are covered in the Tasks and Data track.

A DAG is defined in a Python file placed in Airflow's dags folder. Airflow reads the file and runs its top-level code. Every DAG object that exists when the file ends becomes a DAG Airflow knows about, along with the tasks and edges attached to it. Here is the nightly report as that file, with 4 tasks built from 2 operator types.

3 arguments fix the DAG's identity. dag_id names it, and the name must be unique across the deployment. start_date is the first date the DAG is allowed to run for, and schedule="@daily" asks for one run per day; the Scheduling track covers what both mean in detail. The with DAG(...) as dag: block is the registration: every operator created inside it belongs to this DAG.

An operator is a class that describes one kind of work: run a shell command, call a Python function, send a SQL statement. Creating an instance of it inside the with block, with a task_id, creates one task on the DAG. The file above creates 3 tasks from BashOperator and one from PythonOperator. get_bookings and get_reviews are 2 tasks though they share a class, because each instance has its own task_id and its own command. A task_id must be unique within its DAG. Airflow's record identifies the task by it, and so do the UI and the logs, so the Python variable name report appears nowhere outside this file while build_report appears everywhere.

Airflow reads this file to learn the graph, and it reads it again on a timer to pick up changes. Reading means running the top-level Python: the imports, the function definition, the with block that creates 4 operator instances and 3 edges. The string "python export.py bookings" is stored as a task's argument; the export itself runs later, on a machine Airflow controls, when a run reaches that task. The same split holds for build_report: the function is defined at read time and called at run time. A line at the top of the file that queries the warehouse would run every time Airflow reads the file, dozens of times an hour, against a warehouse that has nothing to do with learning the graph.

>> reads as 'then'. a >> b makes b downstream of a: b may start only after a finishes with success. << is the same edge written from the other side. A list on either end fans out or fans in: [a, b] >> c draws 2 edges into c, and a >> [b, c] draws 2 edges out of a. Chains compose, so the last line of the file draws 3 edges in one expression.

The acyclic rule is enforced at read time. Adding publish >> get_bookings to the file would make the arrows loop, so a run could never finish: publish_dashboard waits for build_report, which waits for get_bookings, which now waits for publish_dashboard. Reading the file raises an error that names the cycle, so the file fails to import: the UI lists it among the DAG import errors, with the cycle in the message, and no graph is loaded from it.

2 tasks may run at the same time when no path of arrows joins them in either direction. get_bookings and get_reviews have no edge between them in nightly_bookings, so when a run begins at 00:00 both start at once. build_report has 2 edges in, so it starts when the later of the 2 finishes. With the bookings export taking 20 minutes and the reviews export 10, the report starts at 00:20. Had the file said get_bookings >> get_reviews >> report, the exports would run one after the other and the report would start at 00:30. The arrows you draw decide the night's length, so draw only the ones the data requires.

The TaskFlow style defines tasks as Python functions. @task on a function turns each call of it, inside a DAG, into a task. @dag on a function turns that function into the DAG definition, and the body of the function is where the tasks are created. Here is the nightly report in this style, with values flowing between the tasks.

No >> appears, and the graph is the same 4 tasks with the same 3 edges. The edges come from the arguments. Passing the result of get_bookings() into build_report tells Airflow that build_report needs a value get_bookings produces, and a task that needs another task's value waits for it: that is the edge. The 2 arguments draw the 2 edges in. Passing report_rows into publish_dashboard draws the third. When get_bookings finishes, Airflow stores its return value, the path string, in its own database; when build_report starts on a worker, Airflow reads that value back and passes it in as bookings_path. The values are serialized as JSON by default, so a value travels if JSON can hold it, which covers strings and numbers as well as lists and dicts. Keep them small, a path, a count, a date, because that database is the record of the run and was sized for records.

The last line, nightly_bookings_flow(), is the one that creates the DAG. @dag turns the function into a factory; calling it runs the body, and each @task call inside creates a task object. Delete that line and the file reads without error and defines no DAG, which is the most common reason a TaskFlow DAG is missing from the UI. Inside the body, get_bookings() runs none of the function's code at read time; it creates the task and returns a reference to the value the task will produce. return "/exports/bookings.csv" executes on a worker, once per run. The print in build_report runs on a worker too, so nothing in these 4 functions touches the process that reads the file.

A task's id defaults to the function name, so build_report the function becomes build_report the task. @task(task_id="aggregate") overrides it. Calling the same decorated function twice creates 2 tasks, and Airflow suffixes the second id: get_bookings and get_bookings__1. 2 calls of one function with different arguments is a common way to get 2 tasks from one definition.

Operators and decorated tasks live in one DAG. Inside the @dag function, publish = BashOperator(task_id="publish_dashboard", bash_command="curl ...") creates a task the same way it would in a with block, and report_rows >> publish orders it after build_report with no value passed, since this publish reads nothing from it. The reverse also works: report_rows can be fed into an operator's templated field. The Tasks and Data track covers how. Use @task where the work is Python you would write anyway; use an operator where a class already wraps the command you need.

The DAG page in the UI has 2 views of the same DAG, switched by a toggle. The graph draws the tasks as boxes and the edges as arrows, laid out left to right, so nightly_bookings shows get_bookings and get_reviews stacked on the left, build_report to their right with 2 arrows in, and publish_dashboard last. The shape alone answers 2 questions: which tasks can overlap, those with no arrow path between them, and what the longest chain is, which bounds how fast a run can finish. Select a run and each box takes the color of that task's state in that run, so the graph becomes a picture of where one night stands.

The grid turns time into columns. Each column is one run, oldest on the left and newest on the right; each row is one task; the cell where they meet is colored by the state of that task in that run. Read down a column to see how one night went; read across a row to see how one task has behaved night after night. 3 nights of bookings_report, the 3-task cousin of nightly_bookings that the animation below plays, make 3 columns of 3 cells. The colors are Airflow's state names, and 7 of them cover everything you will see in this lesson.

Play the night above. At 00:00 on March 2 the first column appears and get_bookings and get_reviews turn lime together. build_report stays blank: the cell has no state until both tasks above it finish. At 00:10 get_reviews turns green and build_report still waits, because one of its 2 edges is unmet. At 00:20 the second edge is met, build_report passes through scheduled and queued and runs, and the column is all green at 00:35. On the night of March 3 build_report turns red at 00:35 and the bar over that column turns red with it: a run fails when a task at the end of its graph fails. Switch the dependencies to the serial file and get_reviews stays blank until get_bookings finishes at 00:20, then runs from 00:20 to 00:30; build_report starts at 00:30 instead of 00:20, so every column finishes 10 minutes later, at 00:45.