11. Deferrable operators and the triggerer

The largest efficiency win available in Airflow, and the one most installations have not taken.

The waste

Chapter 6 named it. A sensor in poke mode holds a worker slot for its entire wait:

   32 worker slots
   30 sensors waiting for files, 6 hours each

   → 2 usable slots
   → tasks pile up in `queued`
   → it looks like a capacity problem
   → someone adds workers
   → the new workers fill with sensors too

mode="reschedule" improves this — the task exits between checks and its slot is released — but it still costs a full task startup every poke, a task_reschedule row every poke (chapter 10), and a scheduler round trip every poke.

Neither approach is right, because waiting is not work. It should not consume a process.

Deferring

A deferrable operator does its setup, works out what it is waiting for, and then hands that wait to the triggerer:

   worker: start the task
           work out the condition
           raise TaskDeferred(trigger=..., method_name="execute_complete")
           → RELEASE THE SLOT ENTIRELY

   triggerer: add this trigger to an asyncio event loop
              alongside thousands of others
              poll asynchronously
              condition met → fire an event

   scheduler: the task is re-queued
   worker:    resume at execute_complete(event)

The task’s state becomes deferred, which is neither running nor queued. It occupies no worker slot and no process — just a row in the database and a coroutine in the triggerer.

   poke mode        1 sensor = 1 worker slot for the whole wait
   reschedule mode  1 sensor = 1 slot briefly, every poke_interval
   DEFERRED         1 sensor = one asyncio coroutine.
                    A triggerer handles THOUSANDS.

Using it

Most modern operators and sensors take deferrable=True:

from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor

S3KeySensor(
    task_id="wait_for_file",
    bucket_key="s3://bucket/data/{{ ds }}.parquet",
    deferrable=True,          # ← that is the whole change
    timeout=60 * 60 * 6,
)

Or turn it on globally:

[operators]
default_deferrable = true

This makes every operator that supports it deferrable without touching a single DAG. On an installation with many sensors this one line is the highest-value configuration change in this guide. Operators that do not support deferral ignore it.

Deferral is not only for sensors. Long-running external jobs benefit just as much:

DatabricksSubmitRunOperator(task_id="etl", deferrable=True, ...)
BigQueryInsertJobOperator(task_id="agg", deferrable=True, ...)
KubernetesPodOperator(task_id="train", deferrable=True, ...)

A Spark job that runs for two hours would otherwise hold a worker for two hours doing nothing but polling. Deferred, it holds nothing.

What a trigger actually is

A small asyncio class with one job: yield an event when a condition is met.

from airflow.triggers.base import BaseTrigger, TriggerEvent
import asyncio

class FileTrigger(BaseTrigger):
    def __init__(self, path: str, poll_interval: float = 30):
        super().__init__()
        self.path, self.poll_interval = path, poll_interval

    def serialize(self):
        # the trigger is STORED IN THE DATABASE — it must round-trip
        return ("plugins.triggers.FileTrigger",
                {"path": self.path, "poll_interval": self.poll_interval})

    async def run(self):
        while True:
            if await check_exists_async(self.path):
                yield TriggerEvent({"status": "success", "path": self.path})
                return
            await asyncio.sleep(self.poll_interval)

And the operator that defers to it:

from airflow.models import BaseOperator
from airflow.exceptions import TaskDeferred

class WaitForFile(BaseOperator):
    def __init__(self, *, path, **kwargs):
        super().__init__(**kwargs)
        self.path = path

    def execute(self, context):
        if file_exists(self.path):       # check once, synchronously
            return self.path             # already there — never defer
        self.defer(trigger=FileTrigger(path=self.path),
                   method_name="execute_complete")

    def execute_complete(self, context, event=None):
        return event["path"]

Four rules that matter when writing one:

serialize() must round-trip. The trigger is stored in the database and reconstructed by the triggerer, which is a different process. Only JSON-serialisable arguments.

run() must be genuinely async. A blocking call — requests.get, time.sleep, a synchronous database driver — blocks the entire triggerer event loop, and with it every other trigger in the process. This is the single biggest mistake in custom triggers, and the symptom is that unrelated deferred tasks stop firing. Use aiohttp, asyncpg, asyncio.sleep.

Check synchronously first. If the condition is already met, return without deferring. Deferring and immediately resuming is pure overhead.

Triggers can fire more than once in edge cases. execute_complete should be idempotent, like everything else in Airflow.

Operating the triggerer

airflow triggerer

[triggerer]
default_capacity = 1000       # triggers per triggerer process

Properties worth knowing:

Run at least two. They are stateless and triggers are redistributed if one dies — but if you run one and it dies, every deferred task hangs until it comes back.

Capacity is high. A thousand concurrent triggers in one process is the default and it is not ambitious; these are coroutines, not threads.

It needs your DAG code, because it must deserialise trigger classes defined in your plugins or DAGs.

A blocked event loop is the failure to watch for. One badly written synchronous trigger stalls everything in that process. triggerer_heartbeat going stale while the process is alive is exactly this.

Metrics:

   airflow_triggers_running          how many are active
   airflow_triggerer_heartbeat       is it alive
   airflow_triggers_blocked_main_thread   ← the one that matters

When not to defer

Very short waits. A condition met in five seconds is not worth the defer-and-resume round trip.

Custom operators without an async client. If the only library available is synchronous, a deferrable version will block the event loop. mode="reschedule" is the correct choice there.

Anything holding local state. Deferring means the process goes away — an open file handle, a temporary directory, a database transaction does not survive. The operator resumes in a different process, and possibly on a different machine.

That last point catches people. execute and execute_complete may run on entirely different workers. Anything they share must be in the trigger’s event payload or in external storage.

The cost/benefit, concretely

   50 DAGs, each waiting ~4 hours for upstream data

   POKE MODE
     50 worker slots occupied for 4 hours
     → a 64-slot cluster has 14 usable slots
     → you buy more workers to run sensors

   RESCHEDULE MODE (poke_interval=60)
     50 × 240 pokes = 12,000 task startups per day
     12,000 task_reschedule rows per day
     12,000 scheduler round trips

   DEFERRABLE
     0 worker slots
     50 coroutines in one triggerer
     50 database rows

The middle column is why reschedule is a half-measure: it fixes the worker occupancy and replaces it with database load. Deferral fixes both.

What to take from this chapter

• Waiting is not work and should not occupy a process. Deferral is the largest efficiency win available in Airflow.
• A deferred task holds no worker slot and no process — a coroutine in the triggerer and a database row. One triggerer handles thousands.
• deferrable=True on most modern operators; default_deferrable = true turns it on installation-wide in one line.
• Deferral is not only for sensors — long external jobs (Spark, BigQuery, Databricks, pods) benefit just as much.
• Writing a trigger: serialize() must round-trip, run() must be genuinely async (a blocking call stalls every trigger in the process), check synchronously first, and make resumption idempotent.
• Run at least two triggerers. They are stateless; one dying with no replacement hangs every deferred task.
• Watch airflow_triggers_blocked_main_thread — that is the badly-written-trigger signal.
• Do not defer very short waits, operators with no async client, or anything holding local state — execute and execute_complete can run on different machines.
• reschedule mode is a half-measure: it frees the worker and replaces the cost with database rows and scheduler round trips.

Next: connections, variables and secrets — where credentials live, and where they should live instead.