Delegating Compute: Airflow Is the Conductor, Not the Orchestra

Here is a task that looks fine and is a bug:

@task
def build_daily_sales():
    df = pd.read_sql("SELECT * FROM raw.orders", warehouse_engine)
    df = df[df.status == "complete"]
    daily = df.groupby(["order_date", "title_id"]).revenue.sum().reset_index()
    daily.to_sql("daily_sales", warehouse_engine, if_exists="replace")

Every line of it is idiomatic Python. It reads clearly. It also drags the bookshop’s entire orders table across the network into an Airflow worker’s RAM, filters it there, groups it there, and pushes the result back — to the same database that could have done all three operations without moving a single byte over the wire. It will work on the sample data. It will work in staging. And it will fall over the first quarter the bookshop has a good year, at which point your options are “give the worker more memory” or “give the worker more memory again.”

But the memory isn’t even the real problem. The real problem is that the transformation logic now lives inside an orchestrator. It can only run when the scheduler runs it. It can only be tested by running a DAG. It can’t be executed by an analyst in a SQL client, or profiled by the DBA, or reused by the reporting job, or ported to a different warehouse. You have taken a piece of business logic and welded it to your scheduling infrastructure, and every future change to either one now has to consider the other.

Airflow is the conductor, not the orchestra. Its job is to decide what runs, in what order, with what retries, and to keep an honest record of what happened. The computation belongs to the tools that were built to compute: the warehouse for set-based work, dbt for the transform layer, Spark for the genuinely big stuff. Every operator in this chapter is a variation on the same shape — start a job somewhere else, and know whether it worked — and once you see it, the whole “how do I process this data in Airflow?” question dissolves. You don’t. You trigger something that does, and you watch it.

The database: SQLExecuteQueryOperator

Start with the most under-used delegation in the whole stack. An enormous amount of what people call “data processing” is just SQL that belongs in the warehouse: a MERGE, a table swap, a CREATE TABLE AS, a stored procedure. It doesn’t need dbt, it doesn’t need Spark, it needs a statement sent to a database and a database left alone to do its job.

In Airflow 3.x the tool for that is SQLExecuteQueryOperator, and if you learned Airflow on 2.x, the first thing to internalize is what happened to the operators you used to reach for.

The per-database operators are gone

PostgresOperator. SnowflakeOperator. MySqlOperator. SqliteOperator. OracleOperator. They’re gone — deprecated through 2.x and removed by the time you’re on a current 3.x provider stack. There is one operator now, and it lives in the common.sql provider:

from airflow.sdk import dag
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator


@dag(schedule="@daily", catchup=False, template_searchpath="/opt/airflow/include/sql")
def bookshop_marts():
    refresh = SQLExecuteQueryOperator(
        task_id="refresh_daily_sales",
        conn_id="warehouse",
        sql="refresh_daily_sales.sql",
        parameters={"run_date": "{{ ds }}"},
        autocommit=True,
    )
    refresh


bookshop_marts()

This is not a rename for the sake of a rename. The conn_id decides the dialect. Point it at a Postgres connection and it speaks Postgres; point it at a Snowflake connection and it speaks Snowflake — same operator, same code, zero changes. The operator resolves the connection, gets the right hook, and hands the SQL to that hook’s DBAPI driver. Migrating the bookshop from Postgres to Snowflake becomes a connection change rather than a find-and-replace across every DAG in the repo. That’s the whole point of the consolidation, and it’s a good one.

sql: strings, lists, and files

The sql parameter takes three shapes, and the third is the one you want.

An inline string is fine for a one-liner. A list of strings runs each statement in order, in one connection — handy for a TRUNCATE then INSERT pair.

A file path is the real answer. SQLExecuteQueryOperator declares .sql in its template_ext, which means passing sql="refresh_daily_sales.sql" makes Airflow read that file and then Jinja-render it before execution. Your SQL lives in a .sql file, where your editor can highlight it, your linter can check it, and a human can read it without wading through Python quoting. template_searchpath on the DAG (see the example above) tells Airflow where to look, beyond the DAG file’s own folder.

The rendered SQL then shows up in the task’s Rendered Template view in the UI, which means when someone asks “what exactly did last Tuesday’s run execute?”, the answer is three clicks away rather than a reconstruction exercise. That alone justifies the pattern.

parameters, and why f-strings are a security bug

Now the part that matters most and gets taught least.

There are two completely different ways a value can get into your SQL, and they are not interchangeable.

Templating — {{ ds }}, {{ params.region }} — is string interpolation. Jinja renders the value into the SQL text, and the database receives a single string containing that value. Whatever the value was, it is now SQL.

Parameters — the parameters= argument — are bound values. They are handed to the DBAPI driver alongside the statement, and the driver passes them to the database as data, out of band, never parsed as SQL. A parameter cannot change the shape of your query. It can only fill a hole.

The difference is invisible until it isn’t. Consider a DAG triggered with a config payload — the bookshop’s ops team can re-run a region:

-- refresh_daily_sales.sql — DANGEROUS
DELETE FROM analytics.daily_sales WHERE region = '{{ dag_run.conf["region"] }}';

Someone triggers that DAG through the REST API (chapter 14) with {"region": "eu'; DROP TABLE analytics.daily_sales; --"} and the rendered SQL is exactly the injection you’d expect. The templating engine did its job perfectly. It interpolated the string it was given. It has no idea it was building SQL.

The fix is to bind:

-- refresh_daily_sales.sql — safe
DELETE FROM analytics.daily_sales WHERE region = %(region)s;
INSERT INTO analytics.daily_sales
SELECT order_date, title_id, sum(revenue)
FROM raw.orders
WHERE order_date = %(run_date)s AND region = %(region)s
GROUP BY 1, 2;

    refresh = SQLExecuteQueryOperator(
        task_id="refresh_daily_sales",
        conn_id="warehouse",
        sql="refresh_daily_sales.sql",
        parameters={"region": "{{ dag_run.conf['region'] }}", "run_date": "{{ ds }}"},
        split_statements=True,
        autocommit=True,
    )

parameters is itself templated — so {{ dag_run.conf['region'] }} still resolves — but the result of that rendering is bound, not interpolated. The driver sends the query and the value separately. The malicious string arrives at the database as a region name that matches nothing, and your table survives.

One nuance worth flagging so it doesn’t bite you: the placeholder syntax is the driver’s, not Airflow’s. psycopg (Postgres) uses %s positionally or %(name)s for named parameters. Some drivers want ?. Some want :name. There is no Airflow-level abstraction over DBAPI’s paramstyle, so the placeholders in your .sql file are coupled to the database it targets — which is the one place the otherwise-lovely conn_id-decides-the-dialect story leaks.

Rule of thumb: template the things that decide which SQL runs; bind the things that are values in it. A table name chosen by environment is a template. A date, an ID, a region, anything that came from a user or a trigger payload, is a parameter. If you’re ever tempted to f-string a value into a query, you are one careless trigger away from an incident.

autocommit, split_statements, return_last, handler

Four more knobs, each of which quietly changes behaviour.

autocommit (default False) decides whether each statement commits as it runs. Left off, your statements run in a transaction that the hook commits at the end — which is what you want for a multi-statement DELETE-then-INSERT that must be atomic. Turn it on for DDL on databases that can’t run it transactionally, and for anything long-running where you’d rather not hold a transaction open. Getting this wrong is how you end up with an empty mart: the DELETE committed, the INSERT failed, and there was no transaction to roll either one back.

split_statements decides whether a single string containing several ;-separated statements is broken up and executed one at a time. Some drivers refuse multiple statements in one execute() call; others silently run only the first. The default varies by hook, which is precisely why you should set it explicitly whenever your .sql file has more than one statement in it. (Passing a Python list of statements always runs them one by one, regardless.)

handler decides what — if anything — comes back. By default the operator uses a handler that fetches all rows, so a SELECT returns a list of tuples. Pass fetch_one_handler and you get a single row. The results are then pushed to XCom, which brings us to the important caveat: with do_xcom_push=False, the operator doesn’t fetch results at all. That’s not just an optimization — it’s the correct setting for a MERGE or a CREATE TABLE AS, where there is no result set and you don’t want the operator groping for one.

return_last (default True) matters only when several statements ran: True gives you the last statement’s results; False gives you a list of every statement’s results. If a downstream task is reading an XCom from a multi-statement task and getting a shape it didn’t expect, this is why.

Data quality, also delegated

While you’re here: common.sql ships a family of check operators that push assertions into the database rather than pulling rows out to assert on.

from airflow.providers.common.sql.operators.sql import (
    SQLColumnCheckOperator,
    SQLTableCheckOperator,
)

check_columns = SQLColumnCheckOperator(
    task_id="check_daily_sales_columns",
    conn_id="warehouse",
    table="analytics.daily_sales",
    column_mapping={
        "revenue": {"min": {"geq_to": 0}, "null_check": {"equal_to": 0}},
        "title_id": {"distinct_check": {"greater_than": 100}},
    },
)

check_table = SQLTableCheckOperator(
    task_id="check_daily_sales_rows",
    conn_id="warehouse",
    table="analytics.daily_sales",
    checks={"row_count_check": {"check_statement": "COUNT(*) > 0"}},
)

Each of these compiles down to SQL that runs in the warehouse and returns a handful of booleans. No dataframe, no row transfer, no worker memory. It’s the same discipline as everything else in this chapter, applied to testing: ask the engine a question, carry home the answer.

When a hook beats an operator

Operators are declarative and that’s usually right. But when you need a value in Python — a count to branch on, an ID to pass downstream, a decision to make — drop to a hook:

from airflow.sdk import task
from airflow.providers.postgres.hooks.postgres import PostgresHook


@task
def assert_load_landed(ds: str):
    # `ds` comes from the run context. A literal "{{ ds }}" here would be sent to
    # Postgres verbatim, match nothing, and this check would "fail" every single day.
    hook = PostgresHook(postgres_conn_id="warehouse")
    (count,) = hook.get_first(
        "SELECT count(*) FROM raw.orders WHERE order_date = %(d)s",
        parameters={"d": ds},
    )
    if count == 0:
        raise ValueError("no orders landed — upstream extract failed")
    return count

Notice what did not happen: no rows were pulled into the worker, no dataframe was built. The count(*) ran in the database; only the integer came back. That’s the thesis in miniature — even when Airflow touches the data, it asks the compute engine a question and carries home the answer.

DbApiHook gives every SQL hook the same vocabulary: get_first for one row, get_records for many, run for statements, insert_rows for a batch write. And there’s a dataframe method — historically get_pandas_df, now generalized to get_df(sql, df_type="pandas" | "polars") in current common.sql releases — which is the most dangerous method in the provider, because it makes the anti-pattern this chapter opened with a one-liner. It has legitimate uses: a small result set, a few hundred rows, something you genuinely need to iterate in Python. It has no business being pointed at a fact table. If you find yourself writing get_df("SELECT * FROM …"), stop and ask what SQL would have done instead.

The hooks also carry the bulk paths, and those are worth knowing because they’re the fast way to move real volume. PostgresHook.copy_expert() drives Postgres’s COPY — a streaming bulk load that is orders of magnitude faster than INSERT-per-row and never materializes the data in Python at all. Every serious warehouse has an equivalent (COPY INTO on Snowflake, LOAD DATA on BigQuery), and the right shape for a big load is always: land the file in object storage, then tell the warehouse to ingest it. Airflow’s task is two lines long and moves zero bytes.

dbt: don’t reimplement the transform layer

Your SQL transforms already have a home. The dbt series builds the bookshop warehouse — staging models, a daily sales mart, tests, snapshots — and dbt is extremely good at exactly that. Airflow’s job is to run it on a schedule and care whether it passed.

The monolith, and its four real problems

The blunt version is a shell-out, and it’s the right first move:

from airflow.sdk import dag, task


@dag(schedule="@daily", catchup=False)
def bookshop_transform():
    @task.bash
    def dbt_build():
        return "cd /opt/dbt/bookshop && dbt build --target prod"

    dbt_build()


bookshop_transform()

One task, one command. dbt’s exit code decides success — if a test fails, dbt build returns non-zero and the task goes red. For a lot of teams that is genuinely all you need, and reaching for something fancier on day one is a mistake.

Live with it for a quarter, though, and four seams open up, each worse than the last.

One task, one log. A red task tells you “dbt broke.” It does not tell you which model. To find out you’re either scrolling a 4,000-line log or parsing run_results.json by hand. The Airflow graph — the thing you built to give you visibility — shows a single box.

No per-model retry. Model 47 of 60 fails on a transient warehouse timeout. You clear the task. dbt reruns all sixty models, including the fifty-nine that were fine, because Airflow’s unit of retry is the task and the task is the whole project. (dbt retry exists and helps, but only if the target/ directory from the failed run is still there — see below.)

dbt deps on a fresh worker. The dbt project’s packages have to exist before dbt build can run. On a containerized or ephemeral worker they don’t, so either you bake them into the image (and now the Airflow image has a dbt dependency tree in it, with all the version-conflict joy that implies) or you run dbt deps every single time, adding a network round-trip to every run.

target/ on an ephemeral filesystem. dbt writes its manifest.json, run_results.json, and compiled SQL into target/. On a Kubernetes worker, that directory dies with the pod. Everything downstream that wants those artifacts — dbt retry, state-based selection (--select state:modified), lineage tooling, the docs site — is looking at a filesystem that no longer exists.

Plus the one that isn’t about failure at all: profiles.yml has to be somewhere, which means warehouse credentials are now a file on the worker rather than an entry in Airflow’s encrypted connection store, and --profiles-dir becomes a string you paste into every command.

Cosmos: a task per model

Cosmos (the astronomer-cosmos package) closes all of it. It parses your dbt project and renders it as an Airflow TaskGroup where every model and every test is its own task, wired in dbt’s own dependency order. You keep writing dbt; Airflow gets a graph it can actually reason about.

from cosmos import DbtTaskGroup, ProjectConfig, ProfileConfig, RenderConfig, LoadMode
from cosmos.profiles import PostgresUserPasswordProfileMapping

profile = ProfileConfig(
    profile_name="bookshop",
    target_name="prod",
    profile_mapping=PostgresUserPasswordProfileMapping(
        conn_id="warehouse",                       # a real Airflow connection
        profile_args={"schema": "analytics"},
    ),
)


@dag(schedule="@daily", catchup=False)
def bookshop_elt():
    extract = ...  # a normal Airflow task

    transform = DbtTaskGroup(
        group_id="dbt",
        project_config=ProjectConfig("/opt/dbt/bookshop"),
        profile_config=profile,
        render_config=RenderConfig(load_method=LoadMode.DBT_MANIFEST),
    )

    notify = ...  # another normal task

    extract >> transform >> notify

Two things are doing the heavy lifting. ProfileConfig solves the credentials problem: a profile mapping builds profiles.yml at runtime from an Airflow connection, so the warehouse password lives in Airflow’s encrypted store (or a secrets backend) and never in a file on disk. And DbtTaskGroup — as opposed to DbtDag — embeds the dbt graph inside a larger DAG, which is the shape you actually want: extract, then dbt, then notify, all in one graph, with the dbt part exploded into real tasks.

The payoff is per-model retry, per-model logs, per-model duration in the Gantt view, and a dependency graph the on-call engineer can read.

The cost is parse time, and it’s the tradeoff that separates a Cosmos demo from a Cosmos deployment. By default Cosmos runs dbt ls to discover models — correct, and slow, and DAG parsing happens constantly. The production answer is LoadMode.DBT_MANIFEST: point Cosmos at a pre-built manifest.json generated in CI whenever the dbt project changes. You trade a little freshness (the manifest is a snapshot) for a lot of parse speed, and if you take one thing from this section, that’s it.

This is a genuinely deep topic — render modes, mapping dbt’s --select onto Airflow, running Cosmos tasks in containers, how dbt’s state and lineage cross the boundary into Airflow’s Assets — and it has its own home. The Orchestrating dbt with Airflow series takes it apart properly. What belongs here is the shape of the decision: shell out when “dbt passed or it didn’t” is a good enough signal; reach for Cosmos when “which model failed, and can I retry just that one” is a question you ask often enough to pay a parse step for.

Spark: the big crunch belongs on a cluster

When the data genuinely outgrows a single machine, the compute moves to Spark and Airflow’s job shrinks to submit and watch.

SparkSubmitOperator

from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator

reprocess = SparkSubmitOperator(
    task_id="reprocess_order_history",
    conn_id="spark_default",
    application="/opt/jobs/reprocess_orders.py",
    application_args=["--date", "{{ ds }}", "--full-refresh"],
    packages="io.delta:delta-spark_2.12:3.2.0",
    py_files="/opt/jobs/bookshop_utils.zip",
    files="/opt/config/reprocess.conf",
    conf={
        "spark.executor.instances": "10",
        "spark.sql.shuffle.partitions": "400",
    },
    executor_memory="8g",
    driver_memory="4g",
)

The dependency-shipping knobs are the ones that bite when you skip them, because Spark runs your code on machines that have never heard of your project:

• packages pulls Maven coordinates onto the cluster’s classpath — how you add Delta, or a JDBC driver, without building a fat jar.
• jars does the same with local jar files.
• py_files ships Python modules (a .zip or .egg) so your job’s import bookshop_utils resolves on the executors, not just on the driver. This is the single most common Spark-on-Airflow failure: it works locally, it works on the driver, and the executors throw ModuleNotFoundError because nobody shipped the module to them.
• files distributes plain files into each executor’s working directory — config, a lookup CSV, a certificate.
• application_args are the arguments your job’s own main parses, and they’re templated, so {{ ds }} flows in as the run date.
• conf sets Spark properties. spark.executor.instances sizes the run from the DAG; spark.sql.shuffle.partitions is the one you’ll actually spend time tuning.

Client vs. cluster, and where the driver lives

Two things decide where Spark actually runs, and they matter more than any of the above.

Deploy mode. In client mode, the Spark driver — the JVM that plans the job, coordinates the executors, and collects results — runs wherever spark-submit was invoked, which is your Airflow worker. In cluster mode, the driver runs out on the cluster in its own container. For anything but the smallest jobs you want cluster: a driver in client mode competes with Airflow for memory, and a driver OOM takes an Airflow worker down with it. (Worse: a df.collect() in client mode pulls the result set into the Airflow worker’s heap, which is the anti-pattern from the top of this chapter, wearing a Spark costume.)

The resource manager — YARN, Kubernetes, or standalone — is set by the master URL in the Spark connection (yarn, k8s://https://…, spark://…). It’s why the operator’s code looks identical whether you’re on a Hadoop cluster or Spark-on-K8s. Deploy mode has historically lived in the connection’s extra alongside spark-home and spark-binary, though recent provider versions surface deploy_mode directly on the operator — check which shape your version wants rather than assuming.

And one cost that’s easy to miss: SparkSubmitOperator shells out to the spark-submit binary. That binary — and a JVM, and the Spark distribution — has to exist on the Airflow worker. You have just given your orchestrator a Java dependency, which is exactly the kind of thing the previous chapter told you to containerize. It also means the operator holds a worker slot for the whole life of the job: it’s a subprocess, not a poll, and there’s no deferrable escape hatch. A six-hour Spark job is a six-hour worker slot.

(There’s a on_kill implementation, at least: kill the Airflow task and the operator does its best to kill the Spark application rather than orphaning it. That’s the discipline from chapter 01, and it’s why you want it.)

What teams actually use instead

Here’s the honest caveat. Most teams don’t run spark-submit from Airflow anymore. If your Spark lives behind a managed service — and it almost certainly does — the operator you reach for is the one that speaks to that service’s API, which means no JVM on the worker, no Spark distribution in the image, and a REST call instead of a subprocess.

Livy — LivyOperator — submits through Apache Livy’s REST endpoint and polls it for completion. It’s the on-prem/self-managed answer, and it’s deferrable, which the raw spark-submit path is not.

from airflow.providers.apache.livy.operators.livy import LivyOperator

reprocess = LivyOperator(
    task_id="reprocess_order_history",
    livy_conn_id="livy_default",
    file="s3://bookshop-jobs/reprocess_orders.py",
    args=["--date", "{{ ds }}"],
    num_executors=10,
    executor_memory="8g",
    polling_interval=30,     # 0 means fire-and-forget — it will NOT wait
    deferrable=True,
)

Note polling_interval. Leave it at its default of 0 and the operator submits the job and succeeds immediately, without waiting for it. Sometimes that’s what you want. Usually it’s a task that goes green while the job it launched is still running — or failing.

EMR — on AWS, EmrAddStepsOperator submits steps to a cluster and returns step IDs; EmrStepSensor waits for one to finish. The submit/sensor split is the classic shape, and both have deferrable modes. EmrServerlessStartJobOperator is the newer, simpler path if you’re not managing a cluster.

Databricks — DatabricksSubmitRunOperator creates and runs a one-off job; DatabricksRunNowOperator triggers a job that already exists in the workspace (which is usually the better boundary: the job is owned and versioned in Databricks, and Airflow just names it and passes parameters). Both poll for completion, both support deferrable=True.

from airflow.providers.databricks.operators.databricks import DatabricksRunNowOperator

reprocess = DatabricksRunNowOperator(
    task_id="reprocess_order_history",
    databricks_conn_id="databricks_default",
    job_id=4471,
    python_params=["--date", "{{ ds }}"],
    idempotency_token="{{ dag.dag_id }}-{{ task.task_id }}-{{ run_id }}",
    deferrable=True,
)

They differ in surface but not in spirit. Airflow hands a job description to a system that owns Spark, and waits. The job runs on ten executors and reprocesses a terabyte of order history; the Airflow task is a thin handle that submits, waits, and reports. Your worker never sees a row.

The pattern, and why it pays

Step back and every operator in this chapter has the same skeleton:

1. Submit a job to an external system.
2. Wait for it — by holding a subprocess, or by polling an API.
3. Interpret the outcome: exit code, job status, run_results.json.
4. Record it, so the graph tells the truth.

That’s it. That’s the whole job. And two things follow from it that are worth making explicit.

Deferrable: the six-hour job that holds zero worker slots

Step 2 is where the money is. A LivyOperator or DatabricksRunNowOperator in classic mode occupies a worker slot for the entire life of the external job — six hours of a worker process asking, every thirty seconds, “done yet?” Sixty concurrent Spark submits pin sixty workers, none of which is computing anything.

deferrable=True hands the waiting to the triggerer (chapter 02): the operator submits, calls self.defer(...), and releases its slot completely. The triggerer’s asyncio loop babysits all sixty polls on one process, and when a job finishes it wakes the task to interpret the result. Sixty six-hour jobs go from sixty pinned workers to zero.

This is the single highest-leverage change available to most Airflow deployments, and it costs one keyword argument. Any operator that waits on an external system should be deferrable if the wait is long or the fan-out is wide. The prerequisite — as always — is that a triggerer is actually running.

Idempotency at the boundary

The uncomfortable truth about submit-and-watch: the submit and the watch can be separated by a failure. Airflow submits a Databricks job, the worker is evicted before it records the run ID, the task is marked for retry, and the retry submits the job again. Now two runs are writing to the same table.

Airflow cannot solve this for you, because the duplicate happened on the other side of the boundary. Which means the boundary is where you solve it, and there are exactly three honest tools:

• An idempotency token, if the platform offers one. Databricks does — idempotency_token in the example above. Submit twice with the same token and Databricks returns the existing run rather than starting a second. Build the token from something stable and unique to the task instance (dag_id + task_id + run_id, as above) and a retry is a no-op instead of a double-run. Use it. It is free correctness.
• Make the job itself idempotent. The job overwrites a partition instead of appending to it. The SQL MERGEs on a key instead of INSERTing. dbt build is idempotent by construction — it rebuilds tables, it doesn’t accumulate into them — which is a large and under-appreciated part of why dbt fits so well under an orchestrator that retries things.
• Check before you submit. A reattach_to_previous_run on the platform’s operator, or a task that queries the external system for an in-flight job with this run’s tag before starting a new one. Clunkier, but it’s what you’re left with when the platform gives you nothing.

If none of those is available, you have an at-least-once boundary and you should know it, because Airflow’s retries — the feature you want — will find it.

Final thoughts

The temptation to compute inside Airflow never really goes away. It’s right there, it’s Python, and for a small job it feels easier than wiring up an external tool. But every one of those tasks is a piece of business logic you have quietly welded to your scheduler, and the bill comes due later: the pandas transform nobody can run outside a DAG, the worker you keep resizing, the transformation nobody can test without a scheduler.

Delegate instead, and the boundaries do real work for you. dbt owns the transforms. The warehouse owns the set-based statements. Spark owns the big crunch. Containers own the awkward code. Airflow owns scheduling, ordering, retries, and the record of what happened. Each of those tools stays independently runnable — you can dbt build by hand, execute the SQL in a client, submit the Spark job from a shell — which means none of them is hostage to your orchestrator, and you can replace any of them without touching the others.

A DAG built this way is one you can read in a minute. It’s a list of things to trigger and the shape of their dependencies, not a pile of business logic wearing an orchestrator as a costume. Push the compute out, keep the control, and defer the waiting — and Airflow becomes what it’s actually good at being: the thing that knows what should have run, whether it did, and what to do when it didn’t.

A conductor who grabs a violin mid-performance isn’t being helpful. They’re just one more thing that can go wrong.

Next: Production Patterns: Idempotent, Observable, Alerting