Idempotent task design in Apache Airflow - Mini Project: Build & Apply

Idempotent Task Design in Airflow

📖 Scenario: You are working with Apache Airflow to automate data processing tasks. To avoid duplicate work and errors, you need to design your tasks so they can run multiple times without causing problems. This is called idempotent task design.Imagine you have a task that writes a file. If the task runs again, it should not create duplicate files or overwrite data incorrectly.

🎯 Goal: Build a simple Airflow DAG with one task that writes a file in an idempotent way. The task should check if the file exists before writing to avoid duplicate writes.

📋 What You'll Learn

Create a Python dictionary called default_args with Airflow DAG default arguments

Create a DAG object called dag with the id idempotent_task_dag

Define a Python function called write_file_if_not_exists that writes to /tmp/idempotent_output.txt only if the file does not exist

Create a PythonOperator task called write_file_task that runs write_file_if_not_exists

Set the task to run inside the DAG

Print a message confirming the file was written or skipped

💡 Why This Matters

🌍 Real World

Idempotent tasks prevent duplicate work and errors in automated workflows, which is critical in data pipelines and system automation.

💼 Career

Understanding idempotent task design is essential for DevOps engineers and data engineers working with workflow automation tools like Apache Airflow.

1

Setup Airflow DAG default arguments

Create a Python dictionary called default_args with these exact entries: 'owner': 'airflow', 'start_date': datetime(2024, 1, 1), and 'retries': 1. Import datetime from the datetime module.

Apache Airflow

from datetime import datetime

# Your code here

Hint

Use default_args = { 'owner': 'airflow', 'start_date': datetime(2024, 1, 1), 'retries': 1 }.

2

Create the Airflow DAG object

Import DAG from airflow. Create a DAG object called dag with the id 'idempotent_task_dag', using the default_args dictionary, and set schedule_interval to '@daily'.

Apache Airflow

from datetime import datetime
from airflow import DAG

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 1, 1),
    'retries': 1
}

# Your code here

Hint

Use dag = DAG('idempotent_task_dag', default_args=default_args, schedule_interval='@daily').

3

Define the idempotent task function

Import os. Define a Python function called write_file_if_not_exists that checks if the file /tmp/idempotent_output.txt exists using os.path.exists. If it does not exist, open the file in write mode and write the text 'This file was created by an idempotent task.'. If the file exists, do nothing.

Apache Airflow

from datetime import datetime
from airflow import DAG
import os

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 1, 1),
    'retries': 1
}

dag = DAG(
    'idempotent_task_dag',
    default_args=default_args,
    schedule_interval='@daily'
)

# Your code here

Hint

Use if not os.path.exists(filepath): and write the file inside that block.

4

Create the PythonOperator and print output

Import PythonOperator from airflow.operators.python. Create a task called write_file_task using PythonOperator with task_id='write_file', python_callable=write_file_if_not_exists, and assign it to the dag. Add a print statement that outputs exactly 'Task write_file completed.'.

Apache Airflow

from datetime import datetime
from airflow import DAG
import os
from airflow.operators.python import PythonOperator

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 1, 1),
    'retries': 1
}

dag = DAG(
    'idempotent_task_dag',
    default_args=default_args,
    schedule_interval='@daily'
)

def write_file_if_not_exists():
    filepath = '/tmp/idempotent_output.txt'
    if not os.path.exists(filepath):
        with open(filepath, 'w') as f:
            f.write('This file was created by an idempotent task.')

# Your code here

Hint

Use write_file_task = PythonOperator(task_id='write_file', python_callable=write_file_if_not_exists, dag=dag) and print('Task write_file completed.').