FileSensor for file arrival detection in Apache Airflow - Step-by-Step Execution

Process Flow - FileSensor for file arrival detection

Start FileSensor Task

↓

Check if file exists

↓

Proceed

↓

Finish

The FileSensor task starts and checks if the target file exists. If yes, it proceeds. If no, it waits and retries until the file arrives or timeout.

Execution Sample

Apache Airflow

from airflow.sensors.filesystem import FileSensor

file_sensor = FileSensor(
    task_id='wait_for_file',
    filepath='/data/input/data.csv',
    poke_interval=5,
    timeout=20
)

This code creates a FileSensor that checks every 5 seconds for the file '/data/input/data.csv' and times out after 20 seconds if the file does not appear.

Process Table

Step	Time (s)	File Exists?	Action	Result
1	0	No	Check file presence	File not found, wait 5s
2	5	No	Check file presence	File not found, wait 5s
3	10	Yes	Check file presence	File found, proceed
4	10	-	Task completes	FileSensor succeeds

💡 File found at 10 seconds, FileSensor task completes successfully

Status Tracker

Variable	Start	After 1	After 2	After 3	Final
file_exists	False	False	False	True	True
time_elapsed	0	5	10	10	10
task_state	running	running	running	success	success

Key Moments - 3 Insights

Why does the FileSensor wait multiple times before succeeding?

Because the file does not exist initially (see execution_table steps 1 and 2), the sensor waits and retries every poke_interval until the file appears or timeout occurs.

What happens if the file never arrives within the timeout?

The FileSensor task will fail after the timeout period (not shown in this trace), stopping retries and marking the task as failed.

Why is the time_elapsed the same at steps 3 and 4?

At step 3, the file is detected, so the task immediately proceeds to completion at step 4 without additional waiting.

Visual Quiz - 3 Questions

Look at the execution table, at which step does the file first exist?

A Step 1

B Step 3

C Step 2

D Step 4

💡 Hint

Check the 'File Exists?' column in the execution_table rows

According to the variable tracker, what is the task_state after step 2?

A running

B failed

C success

D waiting

💡 Hint

Look at the 'task_state' row and the 'After 2' column in variable_tracker

If poke_interval was set to 10 seconds instead of 5, how would the time_elapsed values change in the execution table?

A They would stay the same

B They would increase by 5 seconds each step

C They would increase by 10 seconds each step

D They would decrease

💡 Hint

poke_interval controls wait time between checks, see 'Time (s)' column in execution_table

Concept Snapshot

FileSensor waits for a file to appear before continuing.
It checks the file path every poke_interval seconds.
If the file exists, the task succeeds immediately.
If the file doesn't appear before timeout, the task fails.
Use FileSensor in Airflow to pause workflows until files arrive.

Full Transcript

The FileSensor task in Airflow starts by checking if a specified file exists. If the file is not found, it waits for a set poke_interval before checking again. This repeats until the file appears or the timeout is reached. In the example, the sensor checks every 5 seconds and finds the file at 10 seconds, then completes successfully. Variables like file_exists and task_state update accordingly. If the file never arrives, the task fails after timeout. This sensor helps workflows wait for necessary files before proceeding.

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