Operators: The Verbs of a DAG

A DAG is a sentence about your data. Operators are the verbs — prebuilt task templates for running a command, calling an API, or executing SQL, plus the retry and timeout knobs that make them survive contact with production.

A DAG is the shape of your pipeline — the nouns and the arrows between them. But an arrow doesn’t do anything. Something has to run the command, call the API, execute the query. That something is an operator, and it’s the closest thing Airflow has to a verb.

An operator is a task template. You don’t write a loop that shells out and captures the exit code and handles the timeout and logs the output — someone already did, wrapped it in a class, and called it BashOperator. You hand it the one thing it doesn’t know (the command) and it handles the rest. When the scheduler runs your DAG, each operator instance becomes a task, and each task is a unit of work Airflow can retry, time, and show you in the UI.

The two you’ll reach for first

BashOperator runs a shell command. PythonOperator runs a Python callable. Between them they cover an enormous amount of real work, and they’re the clearest way to see the classic instantiation style — the way DAGs were written before the TaskFlow API, and the way most operators still get used.

from airflow.sdk import DAG
from airflow.providers.standard.operators.bash import BashOperator
from airflow.providers.standard.operators.python import PythonOperator

def load_orders():
    ...

with DAG(dag_id="bookshop_daily", schedule="@daily", catchup=False) as dag:
    fetch = BashOperator(
        task_id="fetch_orders",
        bash_command="curl -sf https://bookshop.internal/orders/$(date +%F).csv "
                     "-o /data/orders/{{ ds }}.csv",
    )

    load = PythonOperator(
        task_id="load_orders",
        python_callable=load_orders,
    )

    fetch >> load

Two things here earn a closer look. The task_id is the task’s name everywhere — the UI, the logs, the retry buttons — so it should read like a verb phrase. And {{ ds }} is a template: Airflow renders it to the run’s logical date before the command executes, so the same DAG fetches a different file every day without you touching the code. Templating is a whole feature of its own, and we’ll come back to it below.

That fetch >> load at the bottom is the dependency: fetch runs, then load runs. In the classic style you draw those arrows yourself.

The knobs that make an operator survive production

Everything so far describes what an operator does the first time it runs cleanly. Production is not the first time it runs cleanly. The network blips, the warehouse is briefly overloaded, a task hangs on a connection that never times out. The reason operators are worth using at all is that every one of them inherits a set of reliability knobs from a common base class, and these knobs — not the bash_command — are what separate a toy DAG from one you’d trust with the nightly load.

Every operator subclasses BaseOperator, and so every operator accepts these arguments. Here’s the fetch task again, dressed for real weather:

from datetime import timedelta

fetch = BashOperator(
    task_id="fetch_orders",
    bash_command="curl -sf https://bookshop.internal/orders/$(date +%F).csv "
                 "-o /data/orders/{{ ds }}.csv",
    retries=3,
    retry_delay=timedelta(minutes=2),
    retry_exponential_backoff=True,
    max_retry_delay=timedelta(minutes=20),
    execution_timeout=timedelta(minutes=10),
)

retries is the one you’ll set most. It’s the number of times Airflow re-runs the task after a failure before giving up. A flaky HTTP endpoint that fails one time in fifty doesn’t deserve to redden your whole DAG — three retries turns that into a non-event. When a task fails but has retries left, its state is up_for_retry, and the scheduler waits before trying again.

How long it waits is retry_delay — a timedelta, two minutes here. Set it long enough that whatever broke has a chance to heal; retrying a rate-limited API after 200 milliseconds just earns you another rate-limit.

retry_exponential_backoff=True changes the delay from constant to doubling: two minutes, then four, then eight. This is the right default for anything talking to a service that might be overloaded, because piling retries onto a struggling system at a fixed interval is how you turn a blip into an outage. max_retry_delay caps that doubling so an unlucky task doesn’t end up waiting hours — here the backoff climbs 2 → 4 → 8 → 16 and then stops at 20 minutes.

execution_timeout is the quiet hero. Without it, a task that hangs — a socket waiting on a server that will never answer — stays running forever, holding a worker slot and blocking everything downstream, and no amount of retries helps because the task never actually fails. With execution_timeout=timedelta(minutes=10), Airflow kills it at the ten-minute mark and marks it failed, which then feeds into your retries. Set this on anything that reaches across the network. A task with no timeout is a task that can wedge your pipeline indefinitely.

Four more knobs matter enough to know by name, even if you set them less often:

depends_on_past=True says this task instance may only run if the same task in the previous DAG run succeeded. Use it for genuinely sequential work — an incremental load where day N’s rows assume day N−1 landed. The catch worth remembering now: if yesterday’s run failed, today’s is blocked until you fix yesterday, which is the point but also a foot-gun if you reach for it reflexively.

trigger_rule decides when a task fires based on its upstream tasks’ states. The default is all_success — run only if every parent succeeded. But you can set all_done (run once every parent has finished, pass or fail — perfect for a cleanup task) or one_failed (run only if a parent failed — perfect for an alert). It’s the mechanism behind “always send the teardown even if the load broke,” and the advanced series takes it apart properly.

pool limits concurrency across DAGs. You define a pool — say warehouse with 5 slots — and every task assigned pool="warehouse" competes for those five slots. It’s how you stop twenty DAGs from opening two hundred connections to a database that tips over at fifty.

Three knobs travel with pool, and they’re worth a moment because they’re the difference between a queue that behaves and one that mysteriously starves:

@task(
    pool="warehouse",
    pool_slots=2,          # this task is heavy: it takes TWO of the five slots
    priority_weight=10,    # ...and when the pool is full, it jumps the queue
    queue="high_memory",   # ...and it must run on a high-memory worker
)
def rebuild_index(): ...

pool_slots (default 1) says how many slots one task instance consumes. The default assumption — every task costs the same — is a lie the moment one of your tasks is a full table rebuild and another is a health check. Give the expensive one pool_slots=2 and the pool stops letting five of them run at once.

priority_weight (default 1) breaks ties when more tasks want a slot than there are slots: higher weight gets scheduled first. The subtlety nobody mentions is that the effective priority isn’t just this number — it’s computed by weight_rule, which defaults to downstream: a task’s weight is its own plus the weight of everything downstream of it. That default is smarter than it looks, because it means a task blocking half the DAG naturally outranks a leaf. Set weight_rule="absolute" when you want the number you wrote to be the number that’s used, and nothing else.

queue routes a task to a particular set of workers — the GPU box, the high-memory box. It does nothing on a single-worker setup and everything once you run more than one worker type, which the executors post covers. Worth knowing now so that when someone says “just put it on the GPU queue,” you know that’s a one-word change and not an architecture.

You will not set all ten on every task. But retries, retry_delay, and execution_timeout belong on nearly everything that touches an external system, and the cheapest way to apply them is once, at the DAG level, via default_args:

default_args = {
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
    "execution_timeout": timedelta(minutes=15),
}

with DAG(dag_id="bookshop_daily", schedule="@daily",
         catchup=False, default_args=default_args) as dag:
    ...

Every task in the DAG inherits those, and any task can override a single one by passing it explicitly. Set the sane baseline once; special-case the exceptions.

Templating: the arguments that render Jinja

Back to {{ ds }}. That double-brace syntax is Jinja, and Airflow renders it just before the task runs, substituting values from the run’s context — ds is the logical date as YYYY-MM-DD, data_interval_start and data_interval_end bound the window, ts is the full timestamp. This is how one static DAG file processes a different slice of data every day.

But templating doesn’t work on every argument — only the ones the operator declares as templatable, listed in its template_fields class attribute. For BashOperator that’s bash_command and env; for SQLExecuteQueryOperator it’s sql, conn_id, and parameters. Pass {{ ds }} into an argument that isn’t in template_fields and you’ll get the literal string {{ ds }}, unrendered — a genuinely confusing first bug, and the fix is always to check whether the field is templatable.

The feature that surprises newcomers: for the fields that support it, Airflow will template the contents of a file, not just an inline string. Both BashOperator and SQLExecuteQueryOperator treat a value ending in .sh or .sql as a path, read the file, and render the Jinja inside it. Combine that with the DAG’s template_searchpath and you keep SQL in .sql files instead of Python strings:

with DAG(
    dag_id="bookshop_daily",
    schedule="@daily",
    catchup=False,
    template_searchpath="/opt/airflow/sql",
) as dag:
    rollup = SQLExecuteQueryOperator(
        task_id="daily_rollup",
        conn_id="bookshop_duckdb",
        sql="daily_rollup.sql",   # read from /opt/airflow/sql/, Jinja rendered
    )

Inside daily_rollup.sql you can write WHERE order_date = '{{ ds }}' and it renders per run. When something looks wrong, open the task instance in the UI and find the Rendered Template view — it shows you exactly what the Jinja resolved to for that specific run, the actual command or SQL that executed. Ninety percent of “why did this task do that?” is answered by reading the rendered template rather than the source.

BashOperator, up close

BashOperator looks trivial and has more sharp edges than any other standard operator. Four are worth internalizing now, because each one has bitten every Airflow user exactly once.

The trailing space. If your bash_command is a template string that ends in .sh, Jinja tries to load it as a file. So bash_command="run_load.sh" reads a file, but bash_command="run_load.sh " — one trailing space — runs the literal command run_load.sh. When you mean “execute this string” and Airflow insists on treating it as a filename, the trailing space is the fix. It’s in the docs and it still catches people.

Exit code 99 means skip. By default BashOperator treats exit code 99 as a signal to skip the task rather than fail it — the state becomes skipped, not failed, and downstream tasks respond accordingly. So exit 99 in your script is a deliberate “nothing to do today, move on cleanly.” You can change the magic number with skip_on_exit_code=, but know that 99 is reserved out of the box, and don’t let a wrapped tool return 99 for an unrelated reason.

cwd, env, and append_env. cwd= sets the working directory the command runs in. env= hands the command a dictionary of environment variables — but be careful: passing env replaces the entire environment, so your command loses PATH and everything else unless you also want that. When you just want to add a couple of variables on top of what’s already there, pass append_env=True alongside env, and Airflow merges instead of replaces:

fetch = BashOperator(
    task_id="fetch_orders",
    bash_command="./scripts/pull.sh",
    cwd="/opt/airflow/bookshop",
    env={"ORDERS_DATE": "{{ ds }}", "BUCKET": "bookshop-raw"},
    append_env=True,
    output_encoding="utf-8",
)

output_encoding controls how Airflow decodes the command’s stdout for the logs — utf-8 is the default and almost always what you want, but a tool that emits Latin-1 will produce mojibake in the logs until you set it to match. Note also that BashOperator pushes the last line of stdout to XCom, so a script that ends by printing a path or an ID hands that value to the next task for free.

PythonOperator, up close

PythonOperator runs a Python callable, and its two most useful arguments pass arguments into that callable. op_args is a list spread as positional arguments; op_kwargs is a dict spread as keyword arguments:

def load_orders(source, *, batch_size, run_date):
    print(f"loading {source} in batches of {batch_size} for {run_date}")

load = PythonOperator(
    task_id="load_orders",
    python_callable=load_orders,
    op_args=["/data/orders"],
    op_kwargs={"batch_size": 500, "run_date": "{{ ds }}"},
)

Both op_args and op_kwargs are templated, which is why "{{ ds }}" renders to the run’s date before your function ever sees it — the string 2026-05-25 arrives as run_date, not the raw template.

Here’s the change from older Airflow that trips up anyone porting a 2.x DAG or copying a Stack Overflow answer: provide_context is gone. In old Airflow you had to pass provide_context=True to make the run context available to your callable. In 3.x the context is injected automatically — you simply declare the context keys you want as parameters and Airflow fills them in:

def load_orders(source, *, data_interval_start=None, ti=None):
    print(f"window starts {data_interval_start}")
    ti.xcom_push(key="row_count", value=1234)

load = PythonOperator(
    task_id="load_orders",
    python_callable=load_orders,
    op_args=["/data/orders"],
)

Name a parameter ti (the task instance), data_interval_start, ds, params — any context key — and it’s there, no flag required. If you still have provide_context=True lying around, delete it; it does nothing and newer versions will complain.

Two more standard operators round out the toolkit, and both are deceptively useful. EmptyOperator does nothing at all — it’s a no-op you use as a structural marker: a single start or finish node that a dozen tasks can fan out from or into, so your graph has a clean join point instead of a tangle. LatestOnlyOperator runs its downstream tasks only for the most recent scheduled run, skipping them during a backfill — handy when part of your DAG (say, refreshing a dashboard) makes no sense to replay for historical dates even though the data-loading part does.

Sensors: operators that wait

There’s a whole category of operator whose entire job is to wait for something to be true before the pipeline proceeds. These are sensors, and the foundations series folds them in here because they’re operators like any other — they just do their work by watching instead of acting.

The shape of the problem: your bookshop load can’t start until an upstream team drops today’s file into a bucket, and that file lands at an unpredictable time each morning. You don’t want to schedule the load at a guessed hour and hope. You want a task that blocks until the file exists, then releases the rest of the DAG. That’s a sensor.

from airflow.providers.standard.sensors.filesystem import FileSensor

wait_for_file = FileSensor(
    task_id="wait_for_orders_file",
    filepath="/data/orders/{{ ds }}.csv",
    poke_interval=60,          # check every 60 seconds
    timeout=60 * 60 * 6,       # give up after 6 hours
    mode="reschedule",
)

The knob that matters most is mode, and it has two settings with very different costs. In poke mode (the default) the sensor holds a worker slot for its entire life, waking every poke_interval seconds to check, then sleeping while still occupying the slot. Fine for a short wait; ruinous if a hundred sensors each sit on a slot for six hours, because you’ve just starved your workers with tasks that are doing nothing but napping.

In reschedule mode the sensor checks, and if the condition isn’t met, it releases the worker slot entirely and asks the scheduler to wake it again after poke_interval. Between checks it costs nothing. The rule of thumb: any wait longer than a minute or two should be mode="reschedule". The trade-off is a little scheduler overhead per check, which is a bargain compared to a wedged worker pool.

timeout is non-negotiable on a sensor. It’s the total time the sensor will wait before failing — six hours here. A sensor with no timeout is a task that can wait forever for a file that’s never coming, and it’ll do it silently. Always cap it.

There’s a third, better option for long waits: deferrable=True. A deferrable sensor hands its wait off to a separate lightweight process called the triggerer, freeing not just the worker slot but nearly all resources while it waits — thousands of deferred sensors cost about what a handful of poking ones do. Many modern sensors support it:

wait_for_file = FileSensor(
    task_id="wait_for_orders_file",
    filepath="/data/orders/{{ ds }}.csv",
    timeout=60 * 60 * 6,
    deferrable=True,
)

The other sensor you’ll meet early is SqlSensor, which polls a query and releases when it returns a truthy first cell — “wait until the staging table has today’s rows before running the transform”:

from airflow.providers.common.sql.sensors.sql import SqlSensor

wait_for_rows = SqlSensor(
    task_id="wait_for_staging",
    conn_id="bookshop_duckdb",
    sql="SELECT count(*) FROM staging_orders WHERE load_date = '{{ ds }}'",
    mode="reschedule",
    poke_interval=120,
    timeout=60 * 60 * 2,
)

This is an introduction, not the last word — the advanced series has a full chapter on sensors, deferral, and the triggerer. For now the mental model is enough: a sensor is an operator that waits, reschedule or deferrable for anything long, and always a timeout.

Finding, installing, and pinning providers

The two Bash and Python operators live in the standard provider, which ships with Airflow. Almost everything else lives in a provider package you install — a versioned pip dependency that bundles operators, hooks, sensors, and connection types for one system. There are over a thousand of them, maintained separately from Airflow’s core so a new Snowflake feature doesn’t have to wait for an Airflow release.

Finding the right one is a search on the Astronomer Registry or the Airflow provider docs for the system you’re integrating — “postgres provider,” “http provider,” “amazon provider.” Installing it is a line in your project’s requirements.txt, which the Astro CLI bakes into the image on the next astro dev restart:

apache-airflow-providers-postgres==6.2.0
apache-airflow-providers-common-sql==1.28.0

Pin the version. An unpinned apache-airflow-providers-postgres will happily install a newer major version on a rebuild months from now, quietly changing an operator’s behavior under a DAG that hasn’t been touched. Providers version independently of Airflow and of each other; pinning is the difference between a reproducible image and a mystery. Each provider also declares which Airflow versions it supports, so check compatibility with your 3.3.0 core before pinning a bleeding-edge release.

SQLExecuteQueryOperator, the one SQL operator

When you want to run a query against Postgres or DuckDB or BigQuery, you don’t write connection-and-cursor boilerplate. In 3.x there’s one unified operator for all of them — SQLExecuteQueryOperator, from the common.sql provider. It replaces the old per-database operators like PostgresOperator, which are deprecated; you point the connection at Postgres or DuckDB, and the same operator class does the rest.

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

rollup = SQLExecuteQueryOperator(
    task_id="daily_rollup",
    conn_id="bookshop_duckdb",
    sql="sql/daily_rollup.sql",
    parameters={"run_date": "{{ ds }}"},
    split_statements=True,
    return_last=True,
    autocommit=True,
)

Each argument earns its place. conn_id points at a connection you define once in the UI or environment — host, credentials, database — so the same operator talks to a different warehouse by pointing at a different connection. sql is either an inline string, a .sql file (rendered through template_searchpath as above), or a list of statements.

parameters is not Jinja templating — it’s the database driver’s own parameter binding, the safe way to inject values so they’re escaped by the driver rather than string-formatted into the query. Your SQL uses the driver’s placeholder (%(run_date)s for many drivers) and the value binds at execution. Reach for parameters over Jinja whenever a value comes from outside, because it’s the injection-safe path.

split_statements=True lets a single .sql file hold several statements separated by semicolons and runs them in order — convenient for a file that creates a table then populates it. return_last=True means the operator pushes only the last statement’s result to XCom (rather than a list of every statement’s result), which is what you want when the final SELECT is the answer and the earlier statements were setup. autocommit=True commits each statement as it runs instead of wrapping the batch in one transaction — appropriate for DDL and idempotent loads, but think twice for a multi-statement change you’d want to roll back as a unit.

The result of that final statement lands in XCom, so a downstream task can read the rollup’s output — a row count, a computed total — and act on it. That hand-off from a SQL operator’s result to your next task is the seam where prebuilt operators and your own glue meet.

So when do you write your own?

The line is cleaner than it looks. Reach for a prebuilt operator when you’re talking to a system someone else has already integrated — a database, a cloud service, an API with a provider. Reach for @task when the work is your logic: parsing a file, reshaping a dict, deciding what to do next. Prebuilt integration versus your own glue.

There’s a rule that applies to both, and it’s the one worth internalizing early: a task should be idempotent. Running it twice with the same inputs should leave the world in the same state as running it once — because Airflow will run it twice. Retries (which you just wired up three of), backfills, a nervous engineer clicking “clear” — all of these re-execute a task, and a task that double-inserts rows or appends to a file each time will quietly corrupt your data. The fetch above overwrites {{ ds }}.csv; run it ten times and you have one correct file. That’s the property to design for, and the reliability knobs above only make sense because of it — retries are safe precisely when the task is idempotent.

Final thoughts

The temptation, when you’re new, is to do everything in PythonOperator — it’s Python, you know Python, why learn a hundred operators? Resist it for anything that touches an external system. A PythonOperator that opens its own database connection is code you now own, test, and debug at 2am; SQLExecuteQueryOperator is code a provider team owns for you, complete with the retry, timeout, and connection handling you’d otherwise reinvent badly. Save your Python for the logic that’s actually yours. The best DAGs are mostly other people’s operators — configured with the reliability knobs turned up — and a little of your glue holding them together.

Next: The TaskFlow API: DAGs That Look Like Python

Comments