When a traditional operator waits for an external event (like an API response or file availability), it holds a worker slot the entire time. This wastes resources because the worker is doing nothing but polling.
Key Insight: A blocking sensor polling every 30 seconds for 1 hour uses only 0.1% of its time productively â the other 99.9% is wasted polling.
How Deferrable Operators Solve This
Aspect
Blocking Operator
Deferrable Operator
Worker Slot
Held during entire wait
Released immediately
Memory Usage
~50-200MB per task
~1KB per trigger
Concurrency
Limited by executor slots
Limited by triggerer capacity
Resource Efficiency
Low
High
Why Deferrable Operators?
Traditional blocking operators hold a worker slot while waiting for external conditions. For long-running operations like API callbacks, database queries, or cloud job completions, this wastes valuable executor resources. Deferrable operators release the worker slot immediately and delegate the wait to the lightweight triggerer process.
For a blocking sensor polling every 30 seconds for 1 hour: (nearly all time is wasted polling). A deferrable equivalent achieves since the worker slot is released immediately.
Creating a Deferrable Operator
from typing import Any, Dict, AsyncIterator
from datetime import timedelta
from airflow.models import BaseOperator
from airflow.triggers.base import BaseTrigger, TriggerEvent
from airflow.utils.decorators import apply_defaults
import asyncio
import aiohttp
class WaitForResultTrigger(BaseTrigger):
"""Async trigger that polls an external API for result completion."""
def __init__(self, poll_url: str, poll_interval: int = 30, token: str = ''):
super().__init__()
self.poll_url = poll_url
self.poll_interval = poll_interval
self.token = token
def serialize(self) -> tuple:
return (
'path.to.WaitForResultTrigger',
{
'poll_url': self.poll_url,
'poll_interval': self.poll_interval,
'token': self.token,
},
)
async def run(self) -> AsyncIterator['TriggerEvent']:
while True:
async with aiohttp.ClientSession() as session:
headers = {'Authorization': f'Bearer {self.token}'}
async with session.get(
self.poll_url, headers=headers
) as response:
data = await response.json()
if data.get('status') == 'completed':
yield TriggerEvent({'status': 'completed', 'result': data})
return
elif data.get('status') == 'failed':
yield TriggerEvent({'status': 'failed', 'error': data.get('error')})
return
await asyncio.sleep(self.poll_interval)
class AsyncJobMonitorOperator(BaseOperator):
"""
Deferrable operator that monitors an external async job.
:param job_url: URL to check job status
:param poll_interval: Seconds between polls
:param execution_timeout: Max time for the entire operation
"""
template_fields = ('job_url',)
@apply_defaults
def __init__(
self,
job_url: str,
poll_interval: int = 30,
execution_timeout: timedelta = timedelta(hours=1),
*args,
**kwargs,
):
super().__init__(*args, execution_timeout=execution_timeout, **kwargs)
self.job_url = job_url
self.poll_interval = poll_interval
def execute(self, context: Dict[str, Any]) -> None:
self.log.info(f'Monitoring job at {self.job_url}')
self.defer(
timeout=self.execution_timeout,
trigger=WaitForResultTrigger(
poll_url=self.job_url,
poll_interval=self.poll_interval,
),
method_name='execute_complete',
)
def execute_complete(self, context: Dict[str, Any], event: Dict[str, Any]) -> Any:
if event['status'] == 'completed':
self.log.info(f'Job completed: {event["result"]}')
return event['result']
else:
raise Exception(f'Job failed: {event.get("error", "Unknown error")}')
HTTP Polling Trigger
from typing import Any, Dict, AsyncIterator
from airflow.triggers.base import BaseTrigger, TriggerEvent
import asyncio
import aiohttp
class HttpPollingTrigger(BaseTrigger):
"""
General-purpose HTTP polling trigger.
Polls a URL until a condition is met or timeout is reached.
"""
def __init__(
self,
url: str,
method: str = 'GET',
headers: Dict[str, str] = None,
expected_status: int = 200,
poll_interval: int = 10,
timeout: int = 300,
):
super().__init__()
self.url = url
self.method = method
self.headers = headers or {}
self.expected_status = expected_status
self.poll_interval = poll_interval
self.timeout = timeout
def serialize(self) -> tuple:
return (
'path.to.HttpPollingTrigger',
{
'url': self.url,
'method': self.method,
'headers': self.headers,
'expected_status': self.expected_status,
'poll_interval': self.poll_interval,
'timeout': self.timeout,
},
)
async def run(self) -> AsyncIterator[TriggerEvent]:
elapsed = 0
while elapsed < self.timeout:
async with aiohttp.ClientSession() as session:
try:
async with session.request(
self.method, self.url, headers=self.headers
) as response:
if response.status == self.expected_status:
data = await response.json()
yield TriggerEvent({
'status': 'success',
'data': data,
})
return
except aiohttp.ClientError:
pass
await asyncio.sleep(self.poll_interval)
elapsed += self.poll_interval
yield TriggerEvent({
'status': 'timeout',
'error': f'Timeout after {self.timeout}s',
})
Triggerer Configuration
# airflow.cfg
[triggerer]
# Number of triggers the triggerer process can run concurrently
capacity = 1000
# How often to check the database for new triggers
triggerer_heartbeat = 5
Key Concepts Table
Component
Purpose
Resource Impact
Scaling
Deferrable Operator
Releases worker slot on defer
Minimal worker usage
Independent of worker count
Trigger Object
Async event definition
Near-zero in memory
Serialized to DB
Triggerer Process
Executes async triggers
Single process, many coroutines
Scale by capacity config
Trigger Callback
Requeues completed task
Metadata DB write
Scales with DB connections
BaseTrigger
Base class for triggers
N/A
N/A
Comparison: Blocking vs Deferrable
Metric
Blocking Operator
Deferrable Operator
Worker slot held
Entire duration
Only during execute()
Memory per wait
~50-200MB (worker process)
~1KB (trigger object)
Concurrency
Limited by executor slots
Limited by triggerer capacity
Latency on completion
Poll interval
Seconds (triggerer heartbeat)
Implementation complexity
Simple
Moderate
Best for
Short waits (<5 min)
Long waits (>5 min)
Code Examples
Custom Trigger with Multiple Conditions
from typing import Any, Dict, AsyncIterator
from airflow.triggers.base import BaseTrigger, TriggerEvent
import asyncio
class MultiConditionTrigger(BaseTrigger):
"""
Trigger that waits for multiple conditions to be satisfied.
"""
def __init__(
self,
conditions: list,
poll_interval: int = 30,
):
super().__init__()
self.conditions = conditions
self.poll_interval = poll_interval
def serialize(self) -> tuple:
return (
'path.to.MultiConditionTrigger',
{
'conditions': self.conditions,
'poll_interval': self.poll_interval,
},
)
async def run(self) -> AsyncIterator[TriggerEvent]:
import aiohttp
while True:
all_met = True
results = {}
async with aiohttp.ClientSession() as session:
for condition in self.conditions:
url = condition['url']
async with session.get(url) as response:
data = await response.json()
met = condition.get('check', lambda d: True)(data)
results[condition['name']] = {
'met': met,
'data': data,
}
if not met:
all_met = False
if all_met:
yield TriggerEvent({'status': 'all_met', 'results': results})
return
self.log.info(
f'Conditions not met: '
f'{[k for k, v in results.items() if not v["met"]]}'
)
await asyncio.sleep(self.poll_interval)
Deferrable Sensor Pattern
from airflow.sensors.base import BaseSensorOperator
from airflow.triggers.base import TriggerEvent
from typing import Any, Dict, AsyncIterator
import asyncio
class DeferrableFileSensor(BaseSensorOperator):
"""
Sensor that defers instead of blocking.
When deferring, the sensor yields a trigger that asynchronously
monitors for file availability without holding a worker slot.
"""
def __init__(self, filepath: str, poll_interval: int = 30, **kwargs):
super().__init__(**kwargs)
self.filepath = filepath
self.poll_interval = poll_interval
def execute(self, context: Dict[str, Any]) -> None:
if not self._check_file():
self.defer(
trigger=FileExistTrigger(self.filepath, self.poll_interval),
method_name='execute_complete',
)
self.log.info(f'File exists: {self.filepath}')
def execute_complete(self, context: Dict[str, Any], event: Dict[str, Any]) -> None:
self.log.info(f'File appeared: {event.get("filepath")}')
def _check_file(self) -> bool:
import os
return os.path.exists(self.filepath)
class FileExistTrigger(BaseTrigger):
def __init__(self, filepath: str, poll_interval: int):
super().__init__()
self.filepath = filepath
self.poll_interval = poll_interval
def serialize(self) -> tuple:
return (
'path.to.FileExistTrigger',
{'filepath': self.filepath, 'poll_interval': self.poll_interval},
)
async def run(self) -> AsyncIterator[TriggerEvent]:
import os
while not os.path.exists(self.filepath):
await asyncio.sleep(self.poll_interval)
yield TriggerEvent({'filepath': self.filepath})
Performance Metrics
Metric
Blocking
Deferrable
Improvement
Worker slot time (1hr wait)
3600s
~0.1s
36000x
Memory per waiting task
~100MB
~1KB
100000x
Max concurrent waits (16 slots)
16
1000+
62x+
Task startup overhead
Worker process
Trigger coroutine
Negligible
Best Practices
Implementation Guidelines
Use for long waits: Deferrable operators shine when tasks wait >5 minutes for external events.
Implement serialize(): Always implement serialize() on triggers for persistence across triggerer restarts.
Handle timeouts: Set execution_timeout on defer calls to prevent infinite waits.
Use async libraries: Leverage aiohttp, httpx, or aiobotocore for non-blocking I/O in triggers.
Operational Guidelines
Monitor triggerer capacity: Track triggers.running and triggers.queued metrics.
Fail gracefully: Handle trigger errors by yielding TriggerEvent({'status': 'error'}).
Avoid blocking in triggers: Never use time.sleep() or synchronous I/O in trigger run() methods.
Test triggers independently: Mock the async loop and verify trigger serialization.