Triggering DAGs with File Sensors

From Time-Based to Event-Based Triggers

In many data engineering scenarios, you don't know exactly when a file will be delivered, only that it will arrive at some point. Running a pipeline every five minutes just to check for a file that might arrive once a day is inefficient. The ideal solution is to have the pipeline wait and only start its work once the data is present. This is the core idea behind event-based triggering.

Airflow achieves this using a special category of operators called Sensors.

What Are Airflow Sensors?

A sensor is an operator whose only job is to wait for a certain condition to be met before succeeding. Once it succeeds, any downstream tasks can begin. This creates a powerful, event-driven workflow. The condition could be many things: a file appearing, a record showing up in a database, a task in another DAG finishing, or a response from an API.

To get a quick and clear overview of what sensors are and why they are so essential, watch the first four minutes of the following video from "Data with Marc".

As you saw, sensors like FileSensor or S3KeySensor are the entry points for many ingestion pipelines. Let's now focus specifically on the FileSensor.

Implementing the FileSensor

The FileSensor is designed to pause a DAG's execution until a file or directory appears at a specified path on the filesystem where the Airflow worker is running. Since our local development setup runs Airflow in a Docker container, this path will be inside the container's filesystem.

The article "How to Implement Airflow Sensors" provides a clear example of how to use it.

The core of the implementation is instantiating the FileSensor operator and providing it with a task ID and the path to the file it should watch.

Here is a typical DAG structure incorporating a FileSensor:

This code snippet shows a complete DAG definition. A FileSensor task with the ID file_sensor is set up to wait for a file before the stop_task can run. Notice the dependency is set with sensor_task >> stop_task.

When this DAG runs, the Airflow UI would show the sensor task in a "running" state while it waits, just like in this diagram:

This shows the 'wait_for_file' FileSensor task actively running (and "poking" for its condition), holding up the downstream 'calculate_sum' PythonOperator task.

Configuring Sensor Behavior: poke_interval and timeout

A sensor doesn't just wait passively; it actively "pokes" or checks its condition at a regular interval. Two parameters are critical for controlling this behavior:

1. poke_interval (in seconds): This determines how often the sensor checks for the condition. The default is 5 seconds. A shorter interval means your pipeline reacts faster but uses more resources.
2. timeout (in seconds): This defines the maximum amount of time the sensor is allowed to run. If the condition is not met within this period, the sensor task will fail.

Setting a timeout is not optional; it is a mandatory best practice. By default, a sensor's timeout is 7 days. If the file never arrives, the sensor task will occupy a worker slot for a full week, which can lead to a critical problem called deadlock, where all your available worker slots are consumed by waiting sensors, preventing any other tasks from running.

The "Data with Marc" video you watched earlier has an excellent segment explaining this danger and the importance of timeout.

Optimizing Resource Usage: Poke vs. Reschedule Mode

As the video highlighted, a sensor in its default poke mode occupies a worker slot for its entire duration, even during the time it's just sleeping between pokes. For sensors that might wait for a long time (minutes or hours), this is highly inefficient.

To solve this, Airflow provides the reschedule mode. When you set mode='reschedule', the sensor task releases its worker slot between pokes. When the poke_interval is up, the task is rescheduled to a worker, performs its check, and if the condition is still not met, it releases the slot again.

This is extremely important for managing resources in a production Airflow environment. The general rule of thumb is:

• Use poke mode (the default) for short waits (less than a few minutes).
• Use reschedule mode for long waits.

The article "How to Implement Airflow Sensors" gives a great summary of this choice.

A Practical Example

Let's integrate this into a practical example. Imagine one of your data sources provides a CSV file named transactions.csv into a directory that is accessible to Airflow (e.g., a mounted volume in your Docker container at /opt/airflow/dags/data).

Here is how you would modify a DAG to wait for this file before processing it.

from __future__ import annotations

import pendulum

from airflow.models.dag import DAG
from airflow.sensors.filesystem import FileSensor
from airflow.operators.python import PythonOperator

def process_file():
    # Your Python logic to read and process the CSV would go here.
    # For now, we'll just print a message.
    print("File has arrived and is now being processed.")

with DAG(
    dag_id="file_triggered_ingestion",
    start_date=pendulum.datetime(2023, 1, 1, tz="UTC"),
    # This DAG can be triggered manually or run on a schedule
    # to 'listen' during a certain window. For file triggers,
    # a daily check is common.
    schedule="@daily",
    catchup=False,
    tags=["ingestion", "sensor"],
) as dag:
    # Task to wait for the file to appear.
    # We will check every 30 seconds and time out after 2 hours.
    # We use 'reschedule' mode because the wait could be long.
    wait_for_new_file = FileSensor(
        task_id="wait_for_new_file",
        filepath="/opt/airflow/dags/data/transactions.csv",
        fs_conn_id="fs_default",  # This is the default connection for the local filesystem
        poke_interval=30,
        timeout=7200, # 2 hours
        mode="reschedule",
    )

    # Task to process the file, which only runs after the sensor succeeds.
    process_the_file = PythonOperator(
        task_id="process_the_file",
        python_callable=process_file,
    )

    wait_for_new_file >> process_the_file

In this DAG, the process_the_file task will not run until the wait_for_new_file sensor task succeeds, which only happens when transactions.csv is found at the specified path.

Conclusion

You've now added a powerful, event-driven pattern to your orchestration toolkit. Instead of guessing when to run your pipeline, you can make it react intelligently to the arrival of data.

Here are the key takeaways from this lesson:

• Airflow Sensors are special operators that pause execution until a condition is met, enabling event-driven workflows.
• The FileSensor waits for a file to appear on the local filesystem.
• Always set a sensible timeout on your sensors to prevent them from running indefinitely and causing deadlocks.
• The poke_interval controls how frequently the sensor checks its condition.
• For any sensor that might wait longer than a few minutes, use mode='reschedule' to release the worker slot between pokes and conserve resources.

In our next lesson, we will combine time-based scheduling with application logic to build an incremental ingestion strategy. You will learn how to use a high-watermark timestamp, stored in your Postgres database, to ensure that each DAG run processes only new or updated data from its specific data interval.