Airflow Sensors: Wait for Files with FileSensor

Creating a DAG with a Sensor

A sensor is a special operator that waits for a condition, checking it repeatedly until it's met.

Every sensor has three key parameters:

• poke_interval: how often the sensor checks, in seconds.
• timeout: how long to wait before failing.
• mode: poke holds the worker slot while waiting (default).

Setting Up the Environment

Run the environment. Once it is up, go to VS Code, open the Airflow directory, then the DAG folder, and create a file called sensor.py.

Writing the Sensor DAG

Import four things: DAG, PythonOperator, datetime, and FileSensor.

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.sensors.filesystem import FileSensor
from datetime import datetime

Define a function that processes the image once it arrives. For now, it only prints a string so you can see something in the logs.

def process_image():
    print("Image processed")

Set up the DAG with dag_id, start_date, schedule, and catchup.

with DAG(
    dag_id="sensor_dag",
    start_date=datetime(2024, 1, 1),
    schedule=None,
    catchup=False
) as dag:

Inside the DAG block, create the sensor task. file_path indicates the file the sensor is watching. poke_interval is set to 10 seconds. timeout is set to 300 seconds (five minutes). mode is set to reschedule to release the worker slot between pokes.

    sensor_task = FileSensor(
        task_id="check_file",
        file_path="/opt/airflow/dags/incoming/photo.jpeg",
        poke_interval=10,
        timeout=300,
        mode="reschedule"
    )

Create a process task using PythonOperator that points to the function.

    process_task = PythonOperator(
        task_id="process_image",
        python_callable=process_image
    )

Set the dependency: the sensor task runs first, then the process task.

    sensor_task >> process_task

Save the file.

Running the DAG

Go to the Airflow UI homepage. Navigate to the DAG section, search for the sensor DAG, open it, trigger it with a single run. You will see that nothing is running yet, and it is up for reschedule because it is looking for the file.

Go back to VS Code, open the DAG folder, create a new folder called incoming, and inside it create an image file called photo.jpeg.

mkdir incoming
touch incoming/photo.jpeg

Go back to the UI. The status is now successful, meaning the DAG found the file and everything worked.