Atomic operations in pipelines in Apache Airflow - Mini Project: Build & Apply

Atomic operations in pipelines

📖 Scenario: You are managing a data pipeline using Apache Airflow. You want to ensure that a critical operation in your pipeline runs atomically, meaning it either completes fully or does not run at all. This prevents partial updates that could cause errors downstream.

🎯 Goal: Build a simple Airflow DAG that performs an atomic operation using a Python function. The DAG should have a task that updates a shared variable only if a condition is met, simulating an atomic update.

📋 What You'll Learn

Create a Python dictionary called shared_data with initial values.

Add a configuration variable update_key to specify which key to update.

Write a Python function atomic_update that updates shared_data atomically if the key exists.

Create an Airflow DAG with a PythonOperator that runs atomic_update.

Print the updated shared_data after the task runs.

💡 Why This Matters

🌍 Real World

In real data pipelines, atomic operations prevent partial updates that can cause inconsistent data or errors. Using Airflow to manage these operations helps keep pipelines reliable and maintainable.

💼 Career

Understanding atomic operations and how to implement them in Airflow is important for data engineers and DevOps professionals who build and maintain robust data workflows.

1

Create initial shared data dictionary

Create a Python dictionary called shared_data with these exact entries: 'counter': 0, 'status': 'pending'.

Apache Airflow

# Create the shared_data dictionary with initial values
# Your code here

Hint

Use curly braces to create a dictionary and include the keys 'counter' and 'status' with the specified values.

2

Add configuration variable for update key

Add a variable called update_key and set it to the string 'counter'.

Apache Airflow

shared_data = {'counter': 0, 'status': 'pending'}
# Add the update_key variable below
# Your code here

Hint

Simply assign the string 'counter' to the variable update_key.

3

Write atomic update function

Write a function called atomic_update that takes no arguments and updates shared_data[update_key] by adding 1 only if update_key exists in shared_data. Use an if statement to check the key.

Apache Airflow

shared_data = {'counter': 0, 'status': 'pending'}
update_key = 'counter'

# Define the atomic_update function below
# Your code here

Hint

Define a function with def. Use if update_key in shared_data: to check the key, then increment the value.

4

Create Airflow DAG and run atomic update

Create an Airflow DAG named atomic_dag with a single PythonOperator task called update_task that runs the atomic_update function. Then, print the shared_data dictionary after the task runs.

Apache Airflow

shared_data = {'counter': 0, 'status': 'pending'}
update_key = 'counter'

def atomic_update():
    if update_key in shared_data:
        shared_data[update_key] += 1

# Import Airflow modules and create DAG with PythonOperator
# Your code here

Hint

Import DAG and PythonOperator. Define the DAG with dag_id='atomic_dag'. Create update_task to run atomic_update. Execute the task and print shared_data.