Suppose an Airflow task starts a long-running Spark job. The computation happens on the Spark cluster, but a standard operator can occupy an Airflow worker slot for the entire wait. With many jobs and few workers, that wait limits how many tasks can run. A deferrable operator releases its worker slot while it waits, leaving the worker available for other tasks. section 9.5.1
Deferral hands the waiting work to Airflow’s triggerer. Its triggers use asynchronous Python code to handle many waits concurrently. When a trigger signals that work is ready, the task returns to a worker to finish. Deferrable operators therefore require at least one triggerer process. section 7.1.2 section 9.5.2
Choose an approach that fits the wait:
- Use a ready-made deferrable operator where available. An `Async` suffix often identifies one, though the suffix is not required; some operators instead offer a `deferrable=True` argument. section 7.1.2
- For a sensor, `mode="reschedule"` releases its task slot between pokes. A deferrable sensor moves the waiting and checks to the triggerer, avoiding a worker for each check. section 7.1.2 section 9.5.2
- For custom waiting logic, implement a deferrable operator and trigger. A custom sensor can inherit from `BaseOperator` rather than `BaseSensorOperator`, because it no longer uses the sensor base class’s `poke` method. section 9.5.2
Wait for ratings before fetching them
A ratings pipeline waits for records to become available before a downstream task fetches them. Its custom sensor calls `self.defer(trigger=MovielensTrigger(...), method_name="execute_complete", timeout=self._timeout)` from `execute`. The trigger defines `serialize` and an asynchronous `run` method. Once the wait ends, `execute_complete` runs on a worker; the DAG can then proceed to the fetch task. The example sets a 60-minute execution timeout. section 9.5.2
Listing 9.32 Custom deferrable sensor
class MovielensSensorAsync(BaseOperator):
def __init__( self, ...):
...
def execute(self, context:Context):
self.defer( #1
trigger=MovielensTrigger( #2
conn_id=self._conn_id,
sleep_interval=self._sleep_interval,
start_date=self._start_date,
end_date=self._end_date,
),
method_name='execute_complete', #3
timeout = self._timeout #4
)
def execute_complete(self, context:Context, event) -> None:
self.log.info(f"Movie Ratings are Available!") #5
See how the custom sensor passes its trigger, completion method, and timeout to `defer`. section 9.5.2
