Airflow Sensors: Waiting for Conditions in Your Pipelines

What Are Sensors?

Sensors are Airflow’s waiting rooms. While regular operators do something — run a script, execute a query, send an email — sensors just wait. They repeatedly check whether a certain condition is true, and only succeed once that condition is met. If the condition is not met within a configurable timeout, the sensor fails the task.

Think of it this way: imagine you work on a data team, and every morning the sales team uploads a CSV file to a shared folder. Your pipeline needs that file before it can start processing. You could schedule your pipeline to run at 10 AM and hope the file is there, but what if the sales team is late one day? A sensor solves this problem elegantly. Instead of guessing when the file will arrive, you tell Airflow, “wait until this file exists, then proceed.”

This pattern comes up constantly in real-world data engineering. Your pipeline might need to wait for a file to land in cloud storage, an API to report that a job is complete, an upstream DAG to finish running, or a database partition to become available. Sensors handle all of these scenarios with the same simple interface: check a condition, and if it is not met yet, check again later.

---

Poke Mode vs Reschedule Mode

Every sensor has a mode parameter that controls how it waits, and choosing the right mode has a significant impact on your cluster’s resource usage. This is one of the most important concepts to understand about sensors.

Poke Mode: Sitting in the Waiting Room

Poke mode is the default, and the name is descriptive — the sensor “pokes” at the condition repeatedly. In this mode, the sensor occupies a worker slot for its entire duration. Between checks, it simply sleeps inside the worker process.

Think of it like sitting in a doctor’s waiting room. You got there, you took a seat, and you are not leaving until the doctor calls your name. You check the clock every few minutes, but you never give up your chair. If the waiting room only has a few chairs (worker slots), you are blocking someone else from sitting down.

from airflow.sensors.filesystem import FileSensor

wait_for_file = FileSensor(
    task_id="wait_for_file",
    filepath="/data/incoming/{{ ds_nodash }}/export.csv",
    mode="poke",
    poke_interval=60,     # Check every 60 seconds
    timeout=3600,         # Give up after 1 hour
)

Poke mode is the right choice when you expect the condition to be met quickly — within a few minutes. The overhead is minimal because the sensor just sleeps between checks without any rescheduling machinery.

Reschedule Mode: Leaving and Coming Back

Reschedule mode takes a different approach. After each failed check, the sensor releases its worker slot entirely. Airflow marks it as “up for rescheduling” and frees the slot for other tasks. When the next poke interval arrives, Airflow re-queues the sensor and it checks again.

Going back to the waiting room analogy: this is like leaving the doctor’s office entirely, going about your day, and coming back in an hour to check again. You are not hogging a chair, but there is overhead each time you walk back in — you need to check in with reception, find a seat, and get your bearings.

wait_for_file = FileSensor(
    task_id="wait_for_file",
    filepath="/data/incoming/{{ ds_nodash }}/export.csv",
    mode="reschedule",
    poke_interval=300,    # Check every 5 minutes
    timeout=7200,         # Give up after 2 hours
)

Reschedule mode is the right choice when you expect long wait times — hours rather than minutes — or when your cluster has limited worker slots. It prevents sensors from starving other tasks of resources. The tradeoff is higher scheduling overhead, so keep the poke_interval above 60 seconds to avoid overwhelming the scheduler.

Quick Comparison

Aspect	Poke	Reschedule
Worker slot usage	Held continuously	Released between checks
Overhead	Low (sleep in process)	Higher (task rescheduled each time)
Best for	Short waits (< 5 min)	Long waits (hours)
Min poke_interval	Any	Should be > 60s

---

FileSensor

The FileSensor waits for a file or directory to appear at a given path. This is one of the most commonly used sensors because so many data pipelines are triggered by file arrivals.

Imagine your pipeline depends on a data export that another team generates daily. They upload it to a shared directory, but the exact time varies — sometimes it arrives at 8 AM, sometimes not until noon. Instead of building brittle scheduling logic, you place a FileSensor at the start of your pipeline and let Airflow handle the waiting.

The filepath parameter supports Jinja templating, so you can parameterize paths based on the execution date. This means the sensor automatically looks for the right file for the right day, even when backfilling historical runs.

from airflow.sensors.filesystem import FileSensor

wait_for_upload = FileSensor(
    task_id="wait_for_upload",
    filepath="/data/uploads/{{ ds_nodash }}/",
    fs_conn_id="fs_default",
    mode="reschedule",
    poke_interval=120,
    timeout=14400,  # 4 hours
)

For files in cloud storage rather than local filesystems, use provider-specific sensors like S3KeySensor (AWS) or GCSObjectExistenceSensor (Google Cloud). They work the same way but check remote storage instead.

---

HttpSensor

The HttpSensor polls an HTTP endpoint and succeeds when a response check function returns True. This is invaluable when your pipeline depends on an external service completing some work.

In practice, many data systems expose status endpoints. A data vendor might provide an API that says whether today’s data feed is ready. A machine learning service might report when a model training run has finished. An internal microservice might signal when it has finished processing a batch. The HttpSensor lets your pipeline wait for any of these conditions without writing polling logic yourself.

You provide a response_check function that examines the HTTP response and returns True when the condition is met. This can be as simple as checking a status field in a JSON response.

from airflow.sensors.http import HttpSensor

wait_for_api = HttpSensor(
    task_id="wait_for_api_ready",
    http_conn_id="data_api",
    endpoint="/status/{{ ds_nodash }}",
    response_check=lambda response: response.json().get("status") == "complete",
    mode="reschedule",
    poke_interval=120,
    timeout=7200,
)

For more complex validation — say you need to check both a status field and a row count — use a named function instead of a lambda. This keeps your code readable and testable.

def validate_data_ready(response):
    data = response.json()
    return data.get("status") == "complete" and data.get("row_count", 0) > 0

wait_for_data = HttpSensor(
    task_id="wait_for_data_ready",
    http_conn_id="data_api",
    endpoint="/datasets/daily_metrics/{{ ds }}",
    response_check=validate_data_ready,
    mode="reschedule",
    poke_interval=120,
    timeout=7200,
)

---

ExternalTaskSensor

The ExternalTaskSensor waits for a task in another DAG to complete. This is how you create dependencies between DAGs without merging them into a single monolithic pipeline.

Think of it this way: your organization has a data ingestion DAG that runs first, loading raw data into the warehouse. Your analytics DAG needs that data to be there before it starts transforming it. These are separate DAGs maintained by separate teams, and you do not want to merge them. The ExternalTaskSensor lets your analytics DAG say, “wait until the ingestion DAG’s final validation task has succeeded for this execution date.”

The key thing to understand is that the sensor matches on execution dates. By default, it looks for the upstream task with the same execution date as the current DAG run. If the upstream DAG runs on a different schedule, you use execution_delta to tell the sensor how to translate between the two schedules.

from airflow.sensors.external_task import ExternalTaskSensor
from datetime import timedelta

wait_for_upstream = ExternalTaskSensor(
    task_id="wait_for_ingestion_dag",
    external_dag_id="data_ingestion",
    external_task_id="final_validation",
    mode="reschedule",
    poke_interval=120,
    timeout=7200,
    execution_delta=timedelta(hours=1),
)

Setting external_task_id to None makes the sensor wait for the entire upstream DAG to complete, not just a single task. The execution_delta parameter in the example above means “look for the upstream DAG run that started one hour before mine.” This is useful when your DAGs run on staggered schedules.

---

Building a Custom Sensor

When none of the built-in sensors fit your use case, you can build your own. Custom sensors extend BaseSensorOperator and implement one method: poke(). This method returns True when the condition is met and False when it is not — Airflow handles everything else.

Imagine your pipeline depends on a specific partition appearing in a database table. Another team’s ETL process creates these partitions, but you do not know exactly when they will finish. A custom sensor can check for the partition’s existence and wait until the data is there.

The beauty of building a custom sensor is that it becomes a reusable component. Instead of writing ad-hoc polling logic in a PythonOperator every time you need to wait for a partition, you create a sensor once and use it across all your DAGs with a clean, declarative interface.

from airflow.sensors.base import BaseSensorOperator
from airflow.hooks.base import BaseHook

class DatabasePartitionSensor(BaseSensorOperator):
    """Waits for a specific partition to exist in a database table."""

    template_fields = ("table", "partition_value")
    ui_color = "#7CB9E8"

    def __init__(self, conn_id, table, partition_column, partition_value, min_rows=1, **kwargs):
        super().__init__(**kwargs)
        self.conn_id = conn_id
        self.table = table
        self.partition_column = partition_column
        self.partition_value = partition_value
        self.min_rows = min_rows

    def poke(self, context):
        hook = BaseHook.get_hook(self.conn_id)
        sql = (
            f"SELECT COUNT(*) FROM {self.table} "
            f"WHERE {self.partition_column} = '{self.partition_value}'"
        )
        result = hook.get_first(sql)

        if result and result[0] >= self.min_rows:
            self.log.info(f"Partition found with {result[0]} rows")
            return True

        self.log.info(f"Partition not ready. Found {result[0] if result else 0} rows, need {self.min_rows}")
        return False

Notice the structure. The __init__ method accepts configuration parameters and the poke method contains the actual check logic. The template_fields tuple enables Jinja templating on table and partition_value, so you can use {{ ds }} in your partition values. Using it in a DAG is clean and self-explanatory.

wait_for_partition = DatabasePartitionSensor(
    task_id="wait_for_orders_partition",
    conn_id="postgres_warehouse",
    table="analytics.orders",
    partition_column="order_date",
    partition_value="{{ ds }}",
    min_rows=100,
    mode="reschedule",
    poke_interval=300,
    timeout=7200,
)

---

Deferrable Operators (Airflow 2.3+)

Both poke mode and reschedule mode have limitations. Poke mode wastes worker slots. Reschedule mode reduces waste but still has scheduling overhead. Starting with Airflow 2.3, deferrable operators offer a third option that uses zero worker slots during the wait.

Here is how it works. When a deferrable sensor checks the condition and finds it is not met, it suspends itself entirely and registers a trigger — a lightweight async coroutine that runs in a separate process called the triggerer. The trigger monitors the condition in the background. When the condition is met, the trigger fires, and Airflow re-queues the task to finish its work.

Think of it this way: instead of sitting in the waiting room (poke) or leaving and coming back (reschedule), you give the reception desk your phone number and go home. They call you when the doctor is ready. Zero chairs occupied, minimal overhead.

Many built-in sensors support deferrable mode out of the box. You just set deferrable=True.

from airflow.sensors.filesystem import FileSensor

wait_for_file = FileSensor(
    task_id="wait_for_file",
    filepath="/data/incoming/{{ ds_nodash }}/export.csv",
    deferrable=True,
    poke_interval=60,
    timeout=14400,
)

Deferrable mode is the best choice when you have many sensors running simultaneously, wait times are long (hours or days), worker slots are a bottleneck, or you are running Airflow 2.3 or later. It does require a running triggerer process in your Airflow deployment, so check with your infrastructure team before relying on it.

---

Timeout and soft_fail

Two parameters that every sensor user should understand are timeout and soft_fail.

The timeout parameter (in seconds) sets the maximum time a sensor will wait before failing. Always set a reasonable timeout. Without one, a sensor could wait forever if the expected condition is never met, turning into a zombie task that consumes resources indefinitely.

The soft_fail parameter changes what happens when a sensor times out. Normally, a timed-out sensor is marked as failed, which can cascade and fail the entire DAG run. With soft_fail=True, the sensor is marked as skipped instead. Downstream tasks in the skip path are also skipped, but the rest of the DAG continues normally.

This is particularly useful for optional dependencies. Imagine your pipeline has an enrichment step that depends on a supplementary data file. If the file arrives, great — use it. If not, skip the enrichment and continue with the rest of the pipeline. That is exactly what soft_fail enables.

wait_for_optional_data = FileSensor(
    task_id="wait_for_optional_enrichment",
    filepath="/data/enrichment/{{ ds_nodash }}.json",
    timeout=1800,
    soft_fail=True,
    mode="reschedule",
    poke_interval=120,
)

---

Next Steps

Sensors give your pipelines the ability to wait for the real world. You now know how to choose between poke mode, reschedule mode, and deferrable mode based on your resource constraints and expected wait times. Here is where to go next:

• Explore the Operators Guide for a deep dive into the operators that run after your sensors succeed
• Learn about XComs to pass data between tasks, including sensor results
• Check out the TaskFlow API for a more Pythonic way to define your pipeline logic
• Always set reasonable timeouts, and consider soft_fail for optional dependencies to keep your pipelines resilient