FileSensor for file arrival detection in Apache Airflow - Commands & Configuration

Introduction

Sometimes you want your workflow to wait until a specific file appears before continuing. FileSensor in Airflow helps by pausing the workflow until the file arrives, so your tasks run only when the file is ready.

When you need to start processing data only after a file is uploaded to a folder.

When your workflow depends on an external system dropping a file before continuing.

When you want to avoid errors by ensuring input files exist before running tasks.

When you want to automate waiting for daily reports to arrive before analysis.

When you want to trigger downstream tasks only after a file transfer completes.

Config File - file_sensor_dag.py

file_sensor_dag.py

from airflow import DAG
from airflow.sensors.filesystem import FileSensor
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta

with DAG(
    dag_id='file_sensor_example',
    start_date=datetime(2024, 1, 1),
    schedule_interval='@daily',
    catchup=False,
    default_args={
        'retries': 1,
        'retry_delay': timedelta(minutes=5),
    },
) as dag:

    wait_for_file = FileSensor(
        task_id='wait_for_file',
        filepath='/tmp/data/input_file.csv',
        poke_interval=10,
        timeout=600,
        mode='poke'
    )

    process_file = BashOperator(
        task_id='process_file',
        bash_command='echo "Processing file..."'
    )

    wait_for_file >> process_file

This DAG waits for the file /tmp/data/input_file.csv to appear before running the processing task.

FileSensor: Watches the file path every 10 seconds (poke_interval) and times out after 10 minutes (timeout).

BashOperator: Runs a simple command after the file is detected.

Task order: The processing task runs only after the file sensor confirms the file exists.

Commands

Lists all available DAGs to confirm your DAG file is recognized by Airflow.

Terminal

airflow dags list

Expected Output

file_sensor_example

example_bash_operator

example_branch_dop_operator

example_xcom

example_subdag_operator

example_trigger_target_dag

Manually triggers the DAG to start the workflow and begin waiting for the file.

Terminal

airflow dags trigger file_sensor_example

Expected Output

Created <DagRun file_sensor_example @ 2024-06-01T12:00:00+00:00: manual__2024-06-01T12:00:00+00:00, externally triggered: True>

Shows the tasks in the DAG to verify the sensor and processing tasks are present.

Terminal

airflow tasks list file_sensor_example

Expected Output

wait_for_file

process_file

Tests the FileSensor task for the given date to see if it waits or completes based on file presence.

Terminal

airflow tasks test file_sensor_example wait_for_file 2024-06-01

Expected Output

INFO - Using executor SequentialExecutor

INFO - Running task instance file_sensor_example.wait_for_file on 2024-06-01

INFO - Poking for file: /tmp/data/input_file.csv

INFO - Task exited with return code 0

Key Concept

If you remember nothing else from this pattern, remember: FileSensor pauses your workflow until the specified file appears, preventing errors from missing inputs.

Common Mistakes

Setting the wrong file path in FileSensor.

The sensor will wait forever or timeout because it never finds the file.

Double-check the exact file path and filename before running the DAG.

Using a very short timeout or poke_interval.

The sensor may timeout too soon or check too frequently, causing unnecessary load.

Set reasonable poke_interval (like 10 seconds) and timeout (like 10 minutes) based on expected file arrival time.

Not setting task dependencies after the sensor.

Downstream tasks may run before the file is ready, causing failures.

Always set downstream tasks to run after the FileSensor task using >> or set_downstream.

Summary

Create a DAG with a FileSensor to wait for a specific file path.

Trigger the DAG and verify tasks to ensure the sensor waits correctly.

Set downstream tasks to run only after the file is detected to avoid errors.

Practice

1. What is the main purpose of using a FileSensor in Airflow?

easy

A. To upload files to a remote server

B. To delete a file after processing

C. To wait until a specific file appears before running the next task

D. To monitor CPU usage during task execution

Solution

1. Step 1: Understand FileSensor's role

   A FileSensor is designed to pause a task until a file is detected at a given path.

2. Step 2: Compare options with FileSensor behavior

   Only To wait until a specific file appears before running the next task describes waiting for a file's arrival, which matches FileSensor's purpose.

3. Final Answer:

   To wait until a specific file appears before running the next task -> Option C

4. Quick Check:

   FileSensor waits for file = A [OK]

Hint: FileSensor pauses until file exists, then continues [OK]

Common Mistakes:

• Confusing FileSensor with file deletion or upload
• Thinking it monitors system resources
• Assuming it triggers immediately without waiting

2. Which of the following is the correct way to create a FileSensor in Airflow to check for a file named data.csv in /tmp directory with poke_interval=30 and timeout=300?

easy

A. FileSensor(task_id='check_file', filepath='/tmp/data.csv', poke_interval=30, timeout=300)

B. FileSensor(task_id='check_file', path='/tmp/data.csv', poke=30, timeout=300)

C. FileSensor(task_id='check_file', filepath='/tmp/data.csv', interval=30, timeout=300)

D. FileSensor(task_id='check_file', filepath='/tmp/data.csv', poke_interval=30)

Solution

1. Step 1: Identify correct parameter names

   The FileSensor uses 'filepath' for the file path, 'poke_interval' for checking frequency, and 'timeout' for max wait time.

2. Step 2: Match options with correct parameters

   FileSensor(task_id='check_file', filepath='/tmp/data.csv', poke_interval=30, timeout=300) uses correct parameter names and values. FileSensor(task_id='check_file', filepath='/tmp/data.csv', poke_interval=30) misses 'timeout', which is optional but often needed.

3. Final Answer:

   FileSensor(task_id='check_file', filepath='/tmp/data.csv', poke_interval=30, timeout=300) -> Option A

4. Quick Check:

   Use 'filepath' and 'poke_interval' exactly [OK]

Hint: Use 'filepath' and 'poke_interval' parameters exactly [OK]

Common Mistakes:

• Using 'path' instead of 'filepath'
• Using 'poke' or 'interval' instead of 'poke_interval'
• Omitting required parameters

3. Given this Airflow task using FileSensor:

file_sensor = FileSensor(
    task_id='wait_for_file',
    filepath='/data/input.txt',
    poke_interval=10,
    timeout=60
)

What will happen if the file /data/input.txt does NOT appear within 60 seconds?

medium

A. The task will skip and continue

B. The task will fail with a timeout error

C. The task will keep waiting indefinitely

D. The task will succeed immediately

Solution

1. Step 1: Understand timeout behavior

   The 'timeout' parameter sets the max seconds the sensor waits before failing.

2. Step 2: Apply to given values

   With timeout=60, if the file doesn't appear in 60 seconds, the task fails with a timeout error.

3. Final Answer:

   The task will fail with a timeout error -> Option B

4. Quick Check:

   Timeout reached = task fails [OK]

Hint: Timeout causes failure if file missing too long [OK]

Common Mistakes:

• Assuming task succeeds without file
• Thinking sensor waits forever ignoring timeout
• Believing task skips automatically on timeout

4. You wrote this FileSensor task:

file_sensor = FileSensor(
    task_id='wait_file',
    filepath='/tmp/data.csv',
    poke_interval='30',
    timeout=300
)

But the task fails immediately with a TypeError. What is the likely cause?

medium

A. task_id cannot contain underscores

B. filepath must be a directory, not a file

C. timeout cannot be set to 300

D. poke_interval should be an integer, not a string

Solution

1. Step 1: Check parameter types

   poke_interval expects an integer number of seconds, not a string.

2. Step 2: Identify error cause

   Setting poke_interval='30' (string) causes TypeError because Airflow expects int.

3. Final Answer:

   poke_interval should be an integer, not a string -> Option D

4. Quick Check:

   poke_interval type must be int [OK]

Hint: poke_interval must be int, not quoted string [OK]

Common Mistakes:

• Passing poke_interval as string instead of int
• Misunderstanding filepath type requirements
• Thinking task_id restrictions cause error

5. You want to create a DAG that waits for multiple files file1.csv and file2.csv to appear in /data before starting processing. Which approach correctly uses FileSensor to achieve this?

hard

A. Create two FileSensor tasks for each file, then set a downstream task that depends on both sensors

B. Create one FileSensor with filepath='/data/file1.csv,/data/file2.csv'

C. Use a single FileSensor with filepath='/data' and wait for directory

D. Create one FileSensor for file1.csv and ignore file2.csv

Solution

1. Step 1: Understand FileSensor scope

   FileSensor waits for one file path; it cannot wait for multiple files at once.

2. Step 2: Design DAG dependencies

   Create two FileSensor tasks, each waiting for one file, then a downstream task that runs after both succeed.

3. Final Answer:

   Create two FileSensor tasks for each file, then set a downstream task that depends on both sensors -> Option A

4. Quick Check:

   Multiple files = multiple sensors + downstream task [OK]

Hint: Use one sensor per file, then join downstream [OK]

Common Mistakes:

• Trying to list multiple files in one filepath
• Waiting on directory instead of files
• Ignoring one of the required files