FileSensor for file arrival detection in Apache Airflow - Mini Project: Build & Apply

FileSensor for file arrival detection

📖 Scenario: You are working with Apache Airflow to automate workflows. You want to create a simple DAG that waits for a specific file to arrive in a folder before proceeding.

🎯 Goal: Build an Airflow DAG that uses FileSensor to detect when a file named data_ready.txt appears in the /tmp/ directory.

📋 What You'll Learn

Create a DAG with the id file_sensor_dag

Use FileSensor to watch for /tmp/data_ready.txt

Set the sensor's poke_interval to 5 seconds

Set the sensor's timeout to 20 seconds

Add a dummy task process_file that runs after the sensor

💡 Why This Matters

🌍 Real World

FileSensors are used in real workflows to wait for data files to arrive before starting processing jobs, ensuring data is ready and avoiding errors.

💼 Career

Understanding sensors in Airflow is important for building reliable data pipelines and automating workflows in data engineering and DevOps roles.

1

Create the DAG and import necessary modules

Write code to import DAG from airflow, FileSensor from airflow.sensors.filesystem, and DummyOperator from airflow.operators.dummy. Then create a DAG object called file_sensor_dag with start_date set to 2024-01-01 and schedule_interval set to @daily.

Apache Airflow

# Your code here

Hint

Use datetime from the datetime module for the start date.

2

Add the FileSensor task to detect the file

Create a FileSensor task called wait_for_file inside the file_sensor_dag that watches for the file /tmp/data_ready.txt. Set poke_interval=5 and timeout=20.

Apache Airflow

from airflow import DAG
from airflow.sensors.filesystem import FileSensor
from airflow.operators.dummy import DummyOperator
from datetime import datetime

dag = DAG(
    dag_id='file_sensor_dag',
    start_date=datetime(2024, 1, 1),
    schedule_interval='@daily'
)

# Your code here

Hint

Remember to pass dag=dag to the task.

3

Add a dummy task to run after the sensor

Create a DummyOperator task called process_file inside the file_sensor_dag. Then set the task dependency so that process_file runs after wait_for_file.

Apache Airflow

from airflow import DAG
from airflow.sensors.filesystem import FileSensor
from airflow.operators.dummy import DummyOperator
from datetime import datetime

dag = DAG(
    dag_id='file_sensor_dag',
    start_date=datetime(2024, 1, 1),
    schedule_interval='@daily'
)

wait_for_file = FileSensor(
    task_id='wait_for_file',
    filepath='/tmp/data_ready.txt',
    poke_interval=5,
    timeout=20,
    dag=dag
)

# Your code here

Hint

Use the >> operator to set task order.

4

Print confirmation of DAG tasks

Add a print statement that outputs the string 'DAG file_sensor_dag with FileSensor and DummyOperator created successfully'.

Apache Airflow

from airflow import DAG
from airflow.sensors.filesystem import FileSensor
from airflow.operators.dummy import DummyOperator
from datetime import datetime

dag = DAG(
    dag_id='file_sensor_dag',
    start_date=datetime(2024, 1, 1),
    schedule_interval='@daily'
)

wait_for_file = FileSensor(
    task_id='wait_for_file',
    filepath='/tmp/data_ready.txt',
    poke_interval=5,
    timeout=20,
    dag=dag
)

process_file = DummyOperator(
    task_id='process_file',
    dag=dag
)

wait_for_file >> process_file

# Your code here

Hint

Use print() to show the confirmation message.

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