Airflow Deferrable Operators: Stop Burning Worker Slots on Sensors

The first Airflow DAG in Building Your First Data Pipeline extracts, transforms, and loads. The next DAG usually waits: for a file on S3, for a partition in the warehouse, for 06:00 UTC, for an upstream DAG to mark success.

Waiting is cheap until it occupies a worker. Then it is slot burn — idle Python processes holding capacity that should be running the load.

flowchart LR
  sensor[Classic sensor] --> slot[Worker slot held]
  slot --> poke[Sleep / poke / sleep]
  poke --> slot
  def[Deferrable sensor] --> yield[Yield trigger]
  yield --> free[Worker freed]
  trig[Triggerer watches]
  trig -->|event| resume[Worker resumes briefly]

Poke holds the slot for the whole wait. Defer holds a triggerer watch and a row in the metadata DB.

Sensors that sleep

A FileSensor in poke mode is a task instance on a worker. It calls poke(), sleeps poke_interval, calls poke() again. The worker slot is occupied the entire time. Forty sensors waiting on forty partner prefixes is forty slots doing nothing useful.

mode='reschedule' is the older workaround: the task releases the slot between pokes and comes back later. It still creates a storm of task-instance state changes. It is better than poke. It is not as cheap as defer.

from airflow.sensors.filesystem import FileSensor

wait_for_file = FileSensor(
    task_id="wait_for_file",
    filepath="/opt/airflow/data/incoming/{{ ds }}.csv",
    poke_interval=60,
    timeout=60 * 60 * 6,
    mode="poke",  # occupies a worker for up to six hours
)

That is fine for a laptop tutorial. It is how a production pool of 32 slots goes idle at 05:00 while every DAG waits on the same SFTP drop.

What deferrable actually does

A deferrable operator reaches a wait, serializes a trigger, and sets the task to deferred. The worker is free. A separate process — the triggerer — runs the trigger (async watch, or a cheap poll that does not hold a worker). When the condition fires, the triggerer sends an event. A worker takes the task again for the short finish path.

You need:

1. A triggerer process (airflow triggerer) in the same deployment as the scheduler.
2. An operator or sensor that implements execute → defer, or exposes deferrable=True.
3. A timeout on the wait. Deferred is not “wait forever in a nicer way.”

No triggerer: tasks go deferred and stay there. The UI looks quiet. The file can be sitting in the bucket for an hour.

from datetime import timedelta

from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from airflow.providers.common.sql.sensors.sql import SqlSensor
from airflow.sensors.date_time import DateTimeSensor

wait_s3 = S3KeySensor(
    task_id="wait_s3",
    bucket_name="lake-landing",
    bucket_key="shop/orders/dt={{ ds }}/_SUCCESS",
    aws_conn_id="aws_default",
    poke_interval=60,
    timeout=60 * 60 * 8,
    deferrable=True,
    mode="reschedule",
)

wait_partition = SqlSensor(
    task_id="wait_orders_partition",
    conn_id="warehouse",
    sql="""
        select 1
        from shop.orders
        where order_date = '{{ ds }}'
        limit 1
    """,
    poke_interval=120,
    timeout=60 * 60 * 6,
    deferrable=True,
)

wait_until_six = DateTimeSensor(
    task_id="wait_until_six_utc",
    target_time="{{ macros.ds_add(ds, 1) }}T06:00:00+00:00",
    poke_interval=30,
    deferrable=True,
)

Provider versions differ on the class name (FileSensorAsync showed up in older docs; current sensors take deferrable=True on the same class). Read the provider you pinned. The habit does not change: the wait must not own a worker.

DateTimeSensor is the cleanest example. Sleeping until 06:00 on a worker is a waste. A trigger that fires at 06:00 is a calendar event.

FileSensor vs deferrable

Local FileSensor on a path the worker can see is a different machine than the triggerer. Deferring a local filesystem wait only works if the triggerer mounts the same path. In Docker Compose that is sometimes true. In Kubernetes it is usually false: the file landed on a worker PVC the triggerer cannot read.

Object storage does not have that problem. S3 / GCS / ADLS sensors talk to an API. The triggerer can poll them. Prefer those for cross-system waits.

# Local disk — poke on the worker that has the volume
from airflow.sensors.filesystem import FileSensor

wait_local = FileSensor(
    task_id="wait_local",
    filepath="/opt/airflow/data/weather/{{ ds }}.json",
    poke_interval=30,
    timeout=60 * 30,
    mode="poke",
)

# Object store — defer; triggerer can see the API
wait_remote = S3KeySensor(
    task_id="wait_remote",
    bucket_name="landing",
    bucket_key="weather/{{ ds }}.json",
    deferrable=True,
    poke_interval=60,
    timeout=60 * 60 * 4,
)

If the extract in the first-pipeline tutorial writes a path and the next DAG waits on it, pass the object-store key, not a worker local path, once you leave a single container.

Slot burn, in numbers you can count

You do not need a vendor chart. Count running task instances whose operator is a sensor and whose duration is hours. That count is slots you cannot use for dbt run or a Spark submit.

Pools make this visible. Put sensors in a small pool if you still have poke-mode leftovers, so they cannot starve the load pool. Then delete the leftovers.

from airflow.decorators import dag, task
from datetime import datetime, timedelta

@dag(
    schedule="@daily",
    start_date=datetime(2026, 1, 1),
    catchup=False,
    max_active_runs=1,
    default_args={"retries": 1, "retry_delay": timedelta(minutes=5)},
)
def orders_when_ready():
    wait = S3KeySensor(
        task_id="wait_success",
        bucket_name="lake-landing",
        bucket_key="shop/orders/dt={{ ds }}/_SUCCESS",
        deferrable=True,
        poke_interval=60,
        timeout=60 * 60 * 8,
        soft_fail=False,
    )

    @task
    def build(ds: str) -> None:
        # invoke dbt or a warehouse merge for that ds
        pass

    wait >> build("{{ ds }}")

orders_when_ready()

soft_fail=True skips downstream on timeout instead of failing. Use it when a missing file is an expected empty day. Do not use it when finance needs that partition — fail and page.

Triggerer operations

Treat the triggerer like the scheduler: one (or a small HA set), healthchecked, with logs you can grep. Capacity is default_capacity (how many triggers one process will run). A triggerer that is overloaded starts lagging events. Sensors look “stuck deferred.”

# compose / systemd / helm — the process must exist
airflow triggerer

In the UI, a deferred task shows the trigger classpath and the timeout. If you see deferred with no triggerer heartbeat, fix the process before you rewrite the DAG.

Triggers should be cheap and idempotent. A trigger that runs a ten-minute warehouse query on every poke interval will just move the slot burn onto the triggerer and the warehouse. SqlSensor should hit a metadata table or a partition directory, not select count(*) from events.

-- cheap existence check the triggerer can afford
select 1
from information_schema.tables
where table_schema = 'shop'
  and table_name = 'orders'
limit 1;

That only proves the table exists. For a daily partition, check a load-audit table your extract already writes:

select 1
from ops.load_audit
where table_name = 'shop.orders'
  and partition_date = '{{ ds }}'
  and status = 'success'
limit 1;

Existence of any row in a 2B-row fact is a full scan if you forget the partition filter. Same pruning habit as any other warehouse query.

flowchart TD
  dag[DAG run] --> wait[Deferrable sensor]
  wait --> def[Task deferred]
  def --> tr[Triggerer]
  tr -->|file or row appears| ev[Trigger event]
  ev --> worker[Worker finishes task]
  worker --> load[Idempotent load for ds]
  tr -.->|timeout| fail[Task failed / soft_fail]

The triggerer owns the wait. The worker owns the write. Timeout is still a first-class path.

When not to defer

• The poke finishes in seconds (a local file the same task just wrote). Defer overhead is silly.
• The wait is on a filesystem only the worker mounts.
• The operator has no deferrable implementation and you are not writing a custom trigger this week. reschedule is acceptable until you replace it.
• You have not deployed a triggerer. Shipping deferrable=True without the process is a silent stall.

Custom operators: if execute() sleeps, you have a sensor in disguise. Either defer or do not call it a wait.

Compare orchestrators in Airflow vs Dagster vs Prefect if you are choosing a platform. Deferrable sensors are an Airflow capacity habit, not a reason to switch.

Migrating a poke DAG

Do not rewrite the load. Change the wait:

1. Confirm a triggerer is running in that environment.
2. Flip deferrable=True on the object-store or SQL sensor. Keep the same timeout and the same ds-keyed path.
3. Run one logical date with airflow dags test (or a manual run). The task should pass through deferred and succeed.
4. Watch worker slot count at the hour you used to stall. It should drop. If it does not, you still have poke-mode sensors in another DAG sharing the pool.

Leave the extract and the delete-then-insert load alone. Defer only fixes who holds the slot. It does not make a non-idempotent load safe.

Pitfalls

• Poke-mode FileSensor on S3 via fuse or a sync sidecar. You still hold a slot. Use the object-store sensor and defer.
• Triggerer not in the deployment. Deferred forever. Healthcheck the process the same way you healthcheck the scheduler.
• Timeout omitted. The wait outlives the business day. Set timeout and decide fail vs soft_fail.
• SqlSensor on an unfiltered fact. You moved the bill from workers to the warehouse. Check an audit table or a partition.
• Local FileSensor deferred in Kubernetes. Triggerer and worker do not share the PVC. The file is “missing” on the wrong box.
• poke_interval of two seconds on a hundred prefixes. The triggerer will poll the API into a rate limit. Minutes are fine for landing files.

Production checklist

• Triggerer runs in every environment that can schedule these DAGs.
• Sensors that wait on S3/GCS/SQL/time are deferrable (or reschedule as a documented exception).
• Local filesystem sensors stay poke-mode on the worker that has the volume — or you stop using local disk as a contract.
• Every wait has timeout and an owner who gets the failure.
• SqlSensor predicates are cheap and partition-pruned.
• Sensor tasks are not in the same pool as heavy compute if any poke-mode sensors remain.
• ds (or the data interval) keys the path or partition. Do not wait on datetime.now().
• Downstream load is idempotent on that interval — defer does not make a retry-safe write.

FAQ

Why are all my workers idle while fifty sensors are “running”? They are poking. Each one owns a slot. Set deferrable=True and run a triggerer, or use mode='reschedule' until you can.

FileSensor vs S3KeySensor — which should I defer? Defer the object-store sensor. Local FileSensor only if the triggerer sees the same path. In Kubernetes, that is usually a no.

What if the triggerer dies mid-wait? Tasks stay deferred until a triggerer resumes and rebuilds watches from the metadata DB. Run it under the same supervision as the scheduler. One triggerer is enough for many teams; HA is a second process, not a second design.

Is reschedule mode good enough? It frees the slot between pokes. It still thrashes task state. Use it as a bridge. Prefer defer for waits measured in hours.

Should I write a custom trigger for a one-off API? Only if you will wait on that API from more than one DAG. A short poke in a dedicated small pool is cheaper than a trigger you will not maintain. If the wait is company-wide (the landing bucket), invest in one deferrable sensor and reuse it.

A first pipeline teaches extract, transform, load, and idempotency. Deferrable sensors teach you not to spend workers on sleep. Deploy the triggerer, defer the long waits, keep timeouts honest, and leave poke-mode for the waits that are actually local and short.