The Parse Loop

Concepts covered: DAG parsing

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.

How often the loop comes back

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.

After an import finishes, the processor looks through the module for DAG objects, serializes each one to JSON and writes the row. The scheduler picks up the new row within seconds. So the clock between saving a file and seeing the change in the UI runs like this: wait for the sweep to reach the file, pay the import, then a few seconds more. On a quiet folder that is under a minute. On a slow folder it is however long a sweep takes, which is the number the rest of this section works out.

What top-level code costs

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.

from airflow.sdk import DAG, task
from warehouse_client import connect

conn = connect("analytics")

with DAG(dag_id="orders_health", schedule=None) as dag:
    @task
    def check_orders():
        rows = conn.execute("SELECT count(*) FROM orders WHERE status = 'open'")
        print(rows.fetchone()[0])

    check_orders()

Say the handshake takes 2 seconds and the folder holds 150 files that all open a connection this way. With 2 processes, one sweep of the folder takes 150 seconds, so each file is imported every 150 seconds rather than every 30, and the folder as a whole completes 1 import per second, around the clock. Each import opens a warehouse session no task will ever use. An edit you save to any file waits for the sweep to reach it, so a one-line fix can take 2 and a half minutes to show in the UI, and the warehouse team sees a steady stream of sessions from a cluster that is running nothing.

• 86,400
• 576

from airflow.sdk import DAG, task

with DAG(dag_id="orders_health", schedule=None) as dag:
    @task
    def check_orders():
        from warehouse_client import connect
        conn = connect("analytics")
        rows = conn.execute("SELECT count(*) FROM orders WHERE status = 'open'")
        print(rows.fetchone()[0])

    check_orders()

The processor is one of 2 places the file is imported. When a worker picks up a task, it imports the file again to find the task's code, so module-level work runs once more per task start, on the worker's host, with the worker's network access. Heavy imports belong inside the task body for the same reason: a numerical library that takes 4 seconds to import costs 4 seconds on every parse, and the processor needs none of it to build the DAG.

Do

• Keep module level to imports from airflow.sdk plus DAG and task construction
• Open connections and clients inside the task function that uses them
• Import heavy libraries inside the task body
• Measure parse time with airflow dags report after each large change

Avoid

• Call any network service at module level
• Read large files or walk directories while the module imports
• Treat a 2-second import as harmless when the folder holds 100 files
• Raise the 50-second timeout before removing the work that hits it

About this lesson

This section is part of the DAG Anatomy: Intermediate lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.