Apache Airflow: Orchestrating Data Pipelines with DAGs

Airflow’s Role in Data Pipelines

Data processing often requires executing a sequence of interdependent tasks. A common pattern involves extracting data, transforming it, and then loading it into a target system. For a single script, a simple Python file or shell script can manage these steps sequentially.

Consider a daily data load:

1. Fetch new records from an API.
2. Clean and normalize the data.
3. Load the processed data into a data warehouse table.

When these operations scale, simple scripts become inadequate. Dependencies between tasks grow complex: task B might require task A to complete successfully, and task C might depend on both A and B. A failure in task A must prevent B and C from starting. Manual intervention for restarts or backfills becomes time-consuming and error-prone.

Scheduling these tasks reliably presents another challenge. Cron jobs can initiate scripts at specific times, but they offer no insight into task status, inter-task dependencies, or error recovery. A cron job simply executes; it does not confirm success, manage retries, or provide a dashboard for monitoring.

Managing task state, retries, and conditional execution across a network of services demands a dedicated system. Without an orchestrator, engineers spend significant effort building custom tooling for monitoring, logging, and dependency management instead of focusing on data logic. This custom tooling often lacks a unified view and consistent failure modes.

Apache Airflow addresses these challenges by providing a programmatic framework to define, schedule, and monitor workflows. It enables defining tasks as directed acyclic graphs (DAGs), where each node is a task and edges define dependencies. This structure ensures tasks execute in the correct order, with built-in mechanisms for retries, error handling, and a centralized UI for operational oversight. Airflow serves as the control plane for complex data movements and transformations, ensuring reliable and observable execution.

DAGs and Tasks: Airflow Workflow Primitives

Airflow orchestrates workflows using Directed Acyclic Graphs (DAGs). A DAG defines a collection of tasks and their dependencies, representing the entire workflow structure without specifying the work itself. This structure ensures a clear, predictable execution path.

The “Directed” aspect means tasks flow in one specific order, from upstream to downstream. “Acyclic” indicates there are no loops; a task cannot depend on itself or on a task that eventually depends on it, which prevents infinite execution cycles. “Graph” refers to the network model where tasks are nodes and dependencies are edges.

Within a DAG, a task represents a single, atomic unit of work. This could be running a Python function, executing a shell command, or transferring data between systems. Tasks are defined, but they do not execute until scheduled by the Airflow scheduler.

Tasks are instantiated from Operators, which are pre-defined templates for common operations. Airflow provides many built-in Operators, such as BashOperator for running shell commands, PythonOperator for calling Python functions, and various Sensor Operators for waiting on external conditions.

Consider a simple workflow that prints messages. This DAG defines three tasks, with a clear execution order.

from airflow.models.dag import DAG
from airflow.operators.bash import BashOperator
from airflow.utils.dates import days_ago

with DAG(
    dag_id='simple_bash_dag',
    start_date=days_ago(1),
    schedule_interval=None,
    catchup=False,
    tags=['example'],
) as dag:
    task_1 = BashOperator(
        task_id='print_hello',
        bash_command='echo "Hello Airflow"',
    )

    task_2 = BashOperator(
        task_id='print_world',
        bash_command='echo "World"',
    )

    task_3 = BashOperator(
        task_id='print_done',
        bash_command='echo "Done"',
    )

    # Define task dependencies
    task_1 >> task_2 >> task_3

In this example, simple_bash_dag is the DAG. print_hello, print_world, and print_done are tasks, each an instance of the BashOperator. The >> operator defines the sequential dependency: task_1 must complete successfully before task_2 starts, and task_2 before task_3.

This structure separates workflow definition (the DAG and its tasks) from the actual execution environment. Airflow manages the scheduling, execution, and monitoring of these defined tasks.

Building a Simple Airflow DAG

Airflow orchestrates workflows by defining Directed Acyclic Graphs (DAGs) in Python files. Each DAG file contains the definition of a workflow, including its tasks and their dependencies. Airflow’s scheduler periodically scans a designated dags folder for these files, parsing them to identify new or updated workflows.

A minimal DAG requires a unique dag_id, a start_date, and a schedule_interval. Tasks within the DAG are defined using operators, which encapsulate specific actions. The BashOperator, for example, executes a shell command.

Consider a simple DAG named hello_world_dag that prints two messages sequentially:

from airflow.models.dag import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

with DAG(
    dag_id="hello_world_dag",
    start_date=datetime(2023, 1, 1),
    schedule_interval=None,
    catchup=False,
    tags=["example"],
) as dag:
    start_task = BashOperator(
        task_id="start_message",
        bash_command='echo "Starting the workflow."',
    )

    end_task = BashOperator(
        task_id="end_message",
        bash_command='echo "Workflow finished."',
    )

    start_task >> end_task

This DAG defines two tasks: start_message and end_message. The >> operator sets a dependency, ensuring end_task runs only after start_task completes successfully. The schedule_interval=None setting indicates this DAG will only run when manually triggered.

Save this Python file, for instance as hello_world_dag.py, into your Airflow dags folder (typically AIRFLOW_HOME/dags). The Airflow scheduler will detect the new file within a few seconds.

Navigate to the Airflow UI in your browser. The hello_world_dag should appear in the DAGs list. Toggle the DAG from “Off” to “On” using the switch. To initiate a run, click the Play icon next to the DAG name and select “Trigger DAG”.

Observe the DAG’s execution in the Graph View. Tasks transition from queued to running and then to success (green) or failed (red). To inspect the output of a specific task, click on it in the Graph View and select “Log”. The logs will show the echo commands’ output, confirming the workflow executed as defined.

Airflow DAGs: Common Pitfalls and Solutions

Airflow DAGs can fail for reasons beyond task logic. Understanding common structural and environment issues accelerates troubleshooting.

Circular Dependencies

The Airflow scheduler cannot resolve task dependencies that form a closed loop. If Task A depends on Task B, and Task B depends on Task A, the DAG will fail to parse or schedule. The scheduler logs will show an airflow.exceptions.AirflowException indicating a cyclical dependency.

Consider this invalid dependency structure:

from airflow.models.dag import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

with DAG(
    dag_id='circular_dag_example',
    start_date=datetime(2023, 1, 1),
    schedule_interval=None,
    catchup=False
) as dag:
    task_a = BashOperator(task_id='task_a', bash_command='echo "Task A"')
    task_b = BashOperator(task_id='task_b', bash_command='echo "Task B"')

    task_a >> task_b
    task_b >> task_a # This creates a cycle

To resolve this, redesign the workflow to remove the circular path. Dependencies must flow in a single direction. For instance, if Task B’s output is an input to Task A, consider a separate task that orchestrates their interaction without a direct dependency loop.

Mutable Default Arguments

default_args is a dictionary often used to set common parameters for all tasks in a DAG. Modifying this dictionary directly after its definition can lead to unintended side effects, as all subsequent tasks will inherit the altered values. Python dictionaries are mutable; changes persist across references.

For example, if you set a retries value for one task by directly modifying default_args, that modification applies to every other task that inherits from the same default_args instance.

from airflow.models.dag import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

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

with DAG(
    dag_id='mutable_args_pitfall',
    default_args=default_args,
    schedule_interval=None,
    catchup=False
) as dag:
    task_1 = BashOperator(task_id='task_1', bash_command='echo "Task 1"')
    
    # Intended for task_2 only, but modifies default_args globally
    default_args['retries'] = 0 
    task_2 = BashOperator(task_id='task_2', bash_command='echo "Task 2"')
    
    # task_1 will now also have retries=0, not 3

To prevent this, create a copy of default_args before making task-specific modifications. Alternatively, pass specific arguments directly to the task constructor, which will override default_args values without altering the original dictionary.

# ... (previous setup)
with DAG(
    dag_id='mutable_args_solution',
    default_args=default_args,
    schedule_interval=None,
    catchup=False
) as dag:
    task_1 = BashOperator(task_id='task_1', bash_command='echo "Task 1"')
    
    # Solution 1: Copy default_args
    task_2_args = default_args.copy()
    task_2_args['retries'] = 0
    task_2 = BashOperator(task_id='task_2', bash_command='echo "Task 2"', **task_2_args)

    # Solution 2: Override directly
    task_3 = BashOperator(task_id='task_3', bash_command='echo "Task 3"', retries=0)

Environment Mismatches

A DAG that runs correctly in a local development environment might fail on an Airflow worker with a ModuleNotFoundError. This typically occurs when Python dependencies are installed locally but are absent from the Airflow worker’s execution environment. The scheduler and workers execute DAG code, so they require access to all necessary libraries.

Ensure all external Python packages used by your DAGs are listed in a requirements.txt file. This file must then be used to install dependencies on every Airflow component (scheduler, workers, webserver) that processes or executes DAG code. For example, if your DAG uses pandas, pandas must be installed in the Airflow environment.

# On the Airflow worker/scheduler host
cd /path/to/airflow/dags_folder
pip install -r requirements.txt

Using containerization technologies like Docker or managed Airflow services helps standardize environments, reducing these discrepancies. The container image or service configuration specifies all required dependencies, ensuring consistency across all Airflow components.

Complex Data Pipeline: Airflow Orchestration Exercise

A common data engineering requirement involves processing raw input through multiple stages, often with dependencies and parallel execution. Consider a daily pipeline that ingests, transforms, loads, and reports on sales data. This scenario demands a multi-task Airflow DAG to ensure correct execution order and handle failures.

The pipeline includes six distinct tasks:

1. ingest_raw_data: Copies a daily sales CSV from a source to a staging area.
2. clean_and_stage: Processes the raw CSV, cleans inconsistencies, and prepares it for loading into a data warehouse.
3. load_to_dwh: Inserts the cleaned data into the target data warehouse table.
4. generate_sales_report: Queries the data warehouse to produce a daily sales summary.
5. generate_inventory_report: Queries the data warehouse to produce an inventory status report.
6. archive_raw_data: Moves the original raw CSV to a long-term archive location after all processing and reporting is complete.

These tasks form a directed acyclic graph. ingest_raw_data must complete before clean_and_stage. clean_and_stage precedes load_to_dwh. Once data is in the DWH, both generate_sales_report and generate_inventory_report can run in parallel. The archive_raw_data task should only execute after both reports have finished.

Define the Python functions for PythonOperator tasks:

import pendulum
from airflow.models.dag import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator

def _clean_and_stage_data():
    """Simulates data cleaning and staging."""
    print("Cleaning and staging raw sales data...")
    # Placeholder for actual data processing logic
    pass

def _generate_sales_report():
    """Simulates generating a daily sales report."""
    print("Generating daily sales report...")
    # Placeholder for report generation logic
    pass

def _generate_inventory_report():
    """Simulates generating an inventory status report."""
    print("Generating daily inventory report...")
    # Placeholder for report generation logic
    pass

The DAG definition combines these tasks with appropriate operators and dependencies:

default_args = {
    "owner": "airflow",
    "depends_on_past": False,
    "email_on_failure": False,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": pendulum.duration(minutes=5),
}

with DAG(
    dag_id="complex_sales_pipeline",
    start_date=pendulum.datetime(2023, 1, 1, tz="UTC"),
    schedule="@daily",
    catchup=False,
    tags=["sales", "data_engineering"],
    default_args=default_args,
) as dag:
    ingest_raw_data = BashOperator(
        task_id="ingest_raw_data",
        bash_command="cp /data/raw/sales_$(date +%Y%m%d).csv /data/staging/sales_raw.csv",
    )

    clean_and_stage = PythonOperator(
        task_id="clean_and_stage_data",
        python_callable=_clean_and_stage_data,
    )

    load_to_dwh = BashOperator(
        task_id="load_to_dwh",
        bash_command="psql -c 'COPY sales FROM /data/staging/sales_cleaned.csv DELIMITER ',' CSV;'"
    )

    generate_sales_report = PythonOperator(
        task_id="generate_sales_report",
        python_callable=_generate_sales_report,
    )

    generate_inventory_report = PythonOperator(
        task_id="generate_inventory_report",
        python_callable=_generate_inventory_report,
    )

    archive_raw_data = BashOperator(
        task_id="archive_raw_data",
        bash_command="mv /data/raw/sales_$(date +%Y%m%d).csv /data/archive/",
    )

    ingest_raw_data >> clean_and_stage >> load_to_dwh
    load_to_dwh >> [generate_sales_report, generate_inventory_report]
    [generate_sales_report, generate_inventory_report] >> archive_raw_data

This DAG structure ensures that data ingestion, cleaning, and loading occur sequentially. Report generation tasks run in parallel after the data is available in the DWH, optimizing execution time. The final archiving step waits for all preceding data processing and reporting to conclude, preventing premature deletion or movement of source files.