Implementing Sensors for Event-Driven Workflow Execution

In this lesson, you will learn how to design sensor tasks to pause your DAG execution until specific external conditions or data availability events are met in upstream systems.

The Concept of Event-Driven Waiting

In robust data pipelines, tasks often depend on files landing in an S3 bucket, a record appearing in a database, or an API status flag flipping to "complete." Instead of using a simple time-based schedule (which is inefficient) or hard-coding dependencies, we use Sensors.

A Sensor is a special type of operator that polls an external system at a set interval. It stays in a running state until the condition it is monitoring evaluates to True. Once the criteria are met, the sensor succeeds, and the downstream tasks in your DAG are triggered. This turns your workflow into an event-driven engine that only proceeds when the data is truly ready for processing.

Practical Implementation

Imagine you are waiting for a daily manifest file to be uploaded to an S3 bucket before triggering your transformation job. Instead of guessing when the file arrives, you implement a S3KeySensor.

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

wait_for_file = S3KeySensor(
    task_id='wait_for_manifest',
    bucket_name='my-data-bucket',
    bucket_key='daily_manifest.csv',
    timeout=60 * 60,  # 1 hour timeout
    poke_interval=60, # Check every minute
    mode='reschedule' # Release the worker slot between checks
)

# Downstream processing task
transform_data = PythonOperator(
    task_id='transform',
    python_callable=process_function
)

wait_for_file >> transform_data

Important Design Consideration: The mode Parameter

Notice the mode='reschedule' parameter. By default, sensors run in poke mode, where they hold an Airflow worker slot open for the entire duration of the wait. This can quickly exhaust your worker capacity if you have many sensors. By using reschedule, the sensor finishes its current execution, releases the worker slot, and sleeps until the next poke_interval. This is crucial for maintaining throughput in busy orchestration environments.

Key Takeaway: Sensors act as smart gates in your workflow, preventing downstream tasks from running until their external prerequisites are satisfied, thereby minimizing wasted compute resources and pipeline failures.