When you're building production data pipelines, especially those consuming data from external APIs, you inevitably hit snags: network flakiness, unexpected API rate limits, or transient processing errors. My team recently faced a situation with a critical financial data pipeline in Nepal where a simple Airflow task retry, without careful design, led to duplicate transaction records and skewed aggregations. Imagine fetching daily stock prices, and a task fails mid-way; a naive retry might refetch the entire day's data, leading to double-counting. This post is for data engineers and developers who are past the basics of Airflow and need to ensure their pipelines are robust and consistent. I'll show you how to build Airflow DAGs that can be rerun safely and predictably, ensuring data integrity regardless of execution outcomes, using a real-world example with JSONPlaceholder Todos.
Key Takeaways
- Implement idempotency in API ingestion tasks to prevent data duplication and ensure consistency during retries.
- Utilize a "high-water mark" pattern to track the last successfully processed record, acting as a persistent cursor for incremental fetches.
- Ensure state updates (like the high-water mark) are transactional, committing them only after new data has been successfully processed and stored.
- Design Airflow tasks to encapsulate these idempotent operations, making your DAGs resilient to transient failures and safe to rerun.
- Simulate failures during development to validate your idempotent logic and observe graceful recovery.
Introduction — The Cost of Non-Idempotent Pipelines
The problem with a simple retry mechanism in Airflow, while often useful, is its inherent assumption that the operation being retried is "safe." If your task involves inserting new records into a database, and it fails *after* some records have been inserted but *before* the task marks itself complete, a retry will simply re-run the entire insertion logic. This can lead to duplicate records, which for financial data—like the daily transaction logs we process from various Nepali banks—is catastrophic. Duplicate entries mean incorrect balances, skewed analytics, and ultimately, distrust in the data. Our previous post on
Scaling Financial Data Scraping in Nepal touched on robust fetching, but it didn't delve into the critical aspect of ensuring data integrity during ingestion, which is where idempotency becomes paramount.
Data and Sources
For this demonstration, we'll be using the JSONPlaceholder Todos API, a free fake API for testing and prototyping. It provides a simple list of todo items, each with a unique ID, `userId`, `title`, and `completed` status. This API is stateless, making it an excellent candidate for demonstrating how we manage state on our end to achieve idempotency.
Data accessed on 2024-07-27.
The Idempotency Imperative: Defining Our Strategy
In the context of data ingestion, an "idempotent" operation means that applying it multiple times produces the same result as applying it once. For fetching data from an external API, this translates to ensuring that if our ingestion task runs, fails, and then runs again, we don't end up with duplicate data or miss any records. Our core strategy here is the "high-water mark" pattern. This involves maintaining a persistent record of the last successfully processed item's identifier (e.g., the highest `id` from the JSONPlaceholder Todos). Each subsequent fetch will only request or process items *newer* than this high-water mark. If a task fails, the high-water mark isn't updated, so on retry, it fetches from the *same* point, effectively resuming where it left off without reprocessing old data.
Step 1 — Setting Up Our Persistent State Store
To implement a high-water mark, we need a place to store its value persistently. For a production system, this would typically be a dedicated database table (e.g., in Postgres or MySQL). For this demonstration, to keep things self-contained and runnable, I'll use a simple JSON file as our mock "database." This file will store our `last_processed_todo_id`.
The sub-problem here is establishing a reliable, albeit simple, mechanism to read and update this critical piece of state.
import json
import os
STATE_FILE = 'pipeline_state.json'
PROCESSED_DATA_FILE = 'processed_todos.jsonl' # JSON Lines for easier appending
def initialize_state():
"""Initializes the state file if it doesn't exist."""
if not os.path.exists(STATE_FILE):
with open(STATE_FILE, 'w') as f:
json.dump({'last_processed_todo_id': 0}, f)
if not os.path.exists(PROCESSED_DATA_FILE):
# Create an empty file if it doesn't exist
open(PROCESSED_DATA_FILE, 'a').close()
def get_high_water_mark():
"""Reads the last_processed_todo_id from the state file."""
initialize_state() # Ensure state file exists before reading
with open(STATE_FILE, 'r') as f:
state = json.load(f)
return state.get('last_processed_todo_id', 0)
def update_high_water_mark(new_hwm):
"""Updates the last_processed_todo_id in the state file."""
with open(STATE_FILE, 'w') as f:
json.dump({'last_processed_todo_id': new_hwm}, f)
# Initialize state on script startup
initialize_state()
This code snippet defines functions to initialize our state file, read the current high-water mark, and update it. `initialize_state()` ensures that if the script runs for the first time, our `pipeline_state.json` and `processed_todos.jsonl` files are created with a sensible default (`last_processed_todo_id: 0`). This means our pipeline will start by fetching all available todos.
Step 2 — Fetching Data Idempotently from JSONPlaceholder
With our state store ready, the next challenge is to fetch data from JSONPlaceholder, but *only* the new data. This is where the high-water mark comes into play. We'll fetch all todos, then filter them based on our `last_processed_todo_id`.
The sub-problem here is efficiently retrieving only the relevant subset of data from a potentially large external API response, without relying on the API itself to filter.
import requests
API_URL = "https://jsonplaceholder.typicode.com/todos"
def fetch_new_todos(current_hwm):
"""
Fetches all todos from the API and filters them based on the high-water mark.
Returns only todos with an ID greater than current_hwm.
"""
print(f"Fetching todos from API. Current high-water mark: {current_hwm}")
try:
response = requests.get(API_URL, timeout=10)
response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx)
all_todos = response.json()
except requests.exceptions.RequestException as e:
print(f"API request failed: {e}")
raise # Re-raise to fail the Airflow task
new_todos = [todo for todo in all_todos if todo['id'] > current_hwm]
print(f"Found {len(new_todos)} new todos.")
return new_todos
The `fetch_new_todos` function first retrieves *all* todos from the API. While some APIs offer server-side filtering (e.g., `?id_gt=X`), JSONPlaceholder does not. So, we perform client-side filtering. This function also includes basic error handling for network issues, raising an exception that Airflow would catch, triggering a retry. Crucially, it only returns items whose `id` is greater than the `current_hwm`.
Step 3 — Processing and Storing Data with Transactional Guarantees
Once we have our `new_todos`, we need to process them and store them in our target system. The "transactional guarantee" here means that our high-water mark should *only* be updated if *all* new todos are successfully processed and stored. If any part of this step fails, the high-water mark should remain unchanged, so a retry will reprocess the same set of `new_todos`.
The sub-problem is ensuring that our persistent state (the high-water mark) reflects the true state of successfully ingested data, even when intermediate processing might fail.
import time
# ... (STATE_FILE, PROCESSED_DATA_FILE, get_high_water_mark, update_high_water_mark, fetch_new_todos functions) ...
# Global flag to simulate failure ONLY ONCE per DAG run for demonstration
# In a real Airflow task, this would likely be handled by a specific task instance failure.
_SIMULATE_FAILURE_TRIGGERED = False
def process_and_store_todos(todos_to_process, current_hwm):
"""
Simulates processing and storing todos.
Updates the high-water mark only if all processing is successful.
Includes a simulated failure point.
"""
global _SIMULATE_FAILURE_TRIGGERED
if not todos_to_process:
print("No new todos to process.")
return current_hwm
max_
