← Back to DevBytes
AI July 11, 2026 15 min read

How to Create a Data Pipeline with Apache Airflow

What is a Data Pipeline and Apache Airflow?

A data pipeline is a series of automated steps that move and transform data from one system to another. Think of it as an assembly line for data — raw data enters at one end, passes through various processing stages such as extraction, cleaning, validation, enrichment, and aggregation, and finally arrives at its destination ready for analysis, reporting, or machine learning. Data pipelines are the backbone of modern data infrastructure, ensuring that data flows reliably and consistently across an organization's systems.

Apache Airflow is an open-source platform designed to programmatically author, schedule, and monitor data pipelines. Originally created at Airbnb in 2014 and later donated to the Apache Software Foundation, Airflow has become the de facto standard for workflow orchestration in the data engineering ecosystem. Unlike simple cron jobs or ad-hoc scripts, Airflow treats pipelines as code, allowing developers to define complex workflows using Python. This brings software engineering best practices — version control, testing, modularity — into the world of data engineering.

At its core, Airflow uses Directed Acyclic Graphs (DAGs) to represent workflows. Each DAG is a collection of tasks with defined dependencies that Airflow executes on a specified schedule. The "directed" part means tasks flow in a specific direction, and "acyclic" ensures there are no circular dependencies that would cause infinite loops. Airflow handles scheduling, retrying failed tasks, sending alerts, and providing a rich web interface for monitoring pipeline health — all while scaling from small personal projects to massive enterprise deployments with thousands of tasks.

Why Apache Airflow Matters

🚀 Deploy your AI agent in 10 minutes

Managed Hermes hosting. Zero DevOps. 100M tokens/mo included.

Try it free →

Before Airflow, data engineers often relied on fragile combinations of cron jobs, custom scripts, and manual intervention to keep data flowing. These approaches had serious limitations:

Airflow solves all of these problems by providing a unified orchestration layer. The benefits are transformative:

Airflow matters because it brings reliability, observability, and engineering rigor to the chaotic world of data movement. Organizations that adopt Airflow report fewer data incidents, faster time-to-insight, and data teams that spend less time firefighting and more time building value.

Core Concepts of Apache Airflow

Understanding Airflow's core concepts is essential before writing your first pipeline. Here are the fundamental building blocks:

DAGs (Directed Acyclic Graphs)

A DAG is the blueprint for your data pipeline. It defines the workflow's structure — which tasks exist, what order they run in, and how they relate to each other. Each DAG has a unique identifier (its dag_id) and properties like schedule interval, start date, and default arguments for tasks. A DAG file is simply a Python script placed in Airflow's DAGs folder, and Airflow parses it to understand your pipeline definition.

Tasks and Operators

A Task is the smallest unit of work in a DAG — it represents a single step in your pipeline. Each task is instantiated from an Operator, which defines what the task actually does. Airflow provides three broad categories of operators:

Task Dependencies

Dependencies define the execution order of tasks. Airflow provides two syntaxes for setting dependencies: the bit-shift operator (>> and <<) and the set_downstream / set_upstream methods. Dependencies can express linear chains, fan-out patterns (one task triggering many), fan-in patterns (many tasks converging into one), and conditional branching using the BranchPythonOperator.

Scheduling and Execution

Airflow's scheduler is the heart of the system. It continuously scans DAG files, determines which tasks are ready to run based on their schedule and dependencies, and dispatches them to workers. Key scheduling concepts include:

Setting Up Apache Airflow

Let's set up a working Airflow environment. For development and learning, the fastest approach is using the official astro-cli or Docker Compose with the Airflow image. Here we'll use a straightforward Docker Compose setup that gives you the full Airflow experience including the scheduler, webserver, and a Postgres backend.

First, create a project directory and a Docker Compose file:

mkdir airflow-tutorial && cd airflow-tutorial

Create the following docker-compose.yml file:

# docker-compose.yml
version: '3.8'

services:
  postgres:
    image: postgres:13
    environment:
      POSTGRES_USER: airflow
      POSTGRES_PASSWORD: airflow
      POSTGRES_DB: airflow
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD", "pg_isready", "-U", "airflow"]
      interval: 10s
      retries: 5

  airflow-init:
    image: apache/airflow:2.8.1
    depends_on:
      postgres:
        condition: service_healthy
    environment:
      AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
    command: db init
    volumes:
      - ./dags:/opt/airflow/dags
      - ./logs:/opt/airflow/logs
      - ./plugins:/opt/airflow/plugins

  scheduler:
    image: apache/airflow:2.8.1
    depends_on:
      airflow-init:
        condition: service_completed_successfully
      postgres:
        condition: service_healthy
    environment:
      AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
      AIRFLOW__CORE__EXECUTOR: LocalExecutor
    command: scheduler
    volumes:
      - ./dags:/opt/airflow/dags
      - ./logs:/opt/airflow/logs
      - ./plugins:/opt/airflow/plugins

  webserver:
    image: apache/airflow:2.8.1
    depends_on:
      airflow-init:
        condition: service_completed_successfully
      postgres:
        condition: service_healthy
    environment:
      AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
      AIRFLOW__CORE__EXECUTOR: LocalExecutor
      AIRFLOW__WEBSERVER__RBAC: 'true'
    command: webserver
    ports:
      - "8080:8080"
    volumes:
      - ./dags:/opt/airflow/dags
      - ./logs:/opt/airflow/logs
      - ./plugins:/opt/airflow/plugins

volumes:
  postgres_data:

Create the required directories and start the services:

mkdir -p dags logs plugins
docker compose up -d

Once the containers are healthy, create an admin user to access the web interface:

docker compose run --rm airflow-init \
  airflow users create \
  --username admin \
  --firstname Admin \
  --lastname User \
  --role Admin \
  --email admin@example.com \
  --password admin

Now navigate to http://localhost:8080 and log in with admin / admin. You should see the Airflow UI with no DAGs yet — we'll fix that next.

Creating Your First Data Pipeline

Now that Airflow is running, let's build a complete data pipeline. This example simulates a common pattern: extracting data from an API, transforming it, and loading it into a database. We'll use Python operators since they're the most flexible and widely used.

Create your first DAG file at dags/weather_pipeline.py:

# dags/weather_pipeline.py
"""
A simple data pipeline that fetches weather data,
transforms it, and stores it in a local JSON file.
"""

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.empty import EmptyOperator
import json
import os
import requests

# Define the default arguments for all tasks in this DAG
default_args = {
    'owner': 'data_team',
    'depends_on_past': False,
    'email_on_failure': True,
    'email': ['alerts@example.com'],
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
    'start_date': datetime(2024, 1, 1),
}

# Instantiate the DAG
with DAG(
    dag_id='weather_data_pipeline',
    default_args=default_args,
    description='Fetch weather data, transform it, and store results',
    schedule_interval='@daily',  # Runs once per day
    catchup=False,
    tags=['tutorial', 'weather'],
) as dag:

    # Task 1: Extract - Fetch data from a weather API
    def fetch_weather_data(**context):
        """Fetch weather data from Open-Meteo free API."""
        # Using Open-Meteo — a free, no-auth-required weather API
        url = (
            "https://api.open-meteo.com/v1/forecast"
            "?latitude=40.7128&longitude=-74.0060"  # New York City
            "&hourly=temperature_2m,relative_humidity_2m,precipitation"
            "&timezone=America/New_York"
        )
        
        response = requests.get(url, timeout=30)
        response.raise_for_status()
        data = response.json()
        
        # Push the raw data to XCom for downstream tasks
        context['task_instance'].xcom_push(
            key='raw_weather_data',
            value=data
        )
        
        print(f"Successfully fetched weather data. "
              f"Records: {len(data.get('hourly', {}).get('time', []))} hours")
    
    fetch_task = PythonOperator(
        task_id='fetch_weather_data',
        python_callable=fetch_weather_data,
        provide_context=True,
    )

    # Task 2: Transform - Clean and structure the data
    def transform_weather_data(**context):
        """Transform raw API response into a structured format."""
        # Pull raw data from XCom
        raw_data = context['task_instance'].xcom_pull(
            task_ids='fetch_weather_data',
            key='raw_weather_data'
        )
        
        if not raw_data:
            raise ValueError("No raw data found in XCom")
        
        hourly = raw_data.get('hourly', {})
        times = hourly.get('time', [])
        temperatures = hourly.get('temperature_2m', [])
        humidities = hourly.get('relative_humidity_2m', [])
        precipitation = hourly.get('precipitation', [])
        
        # Transform into a list of structured records
        transformed_records = []
        for i in range(len(times)):
            record = {
                'timestamp': times[i],
                'temperature_celsius': temperatures[i],
                'humidity_percent': humidities[i],
                'precipitation_mm': precipitation[i],
                'extraction_time': str(datetime.now()),
            }
            transformed_records.append(record)
        
        # Calculate summary statistics
        avg_temp = sum(temperatures) / len(temperatures) if temperatures else 0
        max_temp = max(temperatures) if temperatures else 0
        min_temp = min(temperatures) if temperatures else 0
        
        summary = {
            'num_records': len(transformed_records),
            'avg_temperature': round(avg_temp, 2),
            'max_temperature': max_temp,
            'min_temperature': min_temp,
            'processed_at': str(datetime.now()),
        }
        
        # Push transformed data to XCom
        context['task_instance'].xcom_push(
            key='transformed_records',
            value=transformed_records
        )
        context['task_instance'].xcom_push(
            key='summary',
            value=summary
        )
        
        print(f"Transformed {len(transformed_records)} records")
        print(f"Summary: Avg Temp={avg_temp:.1f}°C, "
              f"Range={min_temp:.1f}°C to {max_temp:.1f}°C")
    
    transform_task = PythonOperator(
        task_id='transform_weather_data',
        python_callable=transform_weather_data,
        provide_context=True,
    )

    # Task 3: Load - Save results to a JSON file
    def load_weather_data(**context):
        """Load transformed data to a local JSON file."""
        records = context['task_instance'].xcom_pull(
            task_ids='transform_weather_data',
            key='transformed_records'
        )
        summary = context['task_instance'].xcom_pull(
            task_ids='transform_weather_data',
            key='summary'
        )
        
        if not records:
            raise ValueError("No transformed records found")
        
        # Create output directory if it doesn't exist
        output_dir = '/opt/airflow/output/weather'
        os.makedirs(output_dir, exist_ok=True)
        
        # Save detailed records
        execution_date = context['execution_date']
        filename = f"weather_{execution_date.strftime('%Y%m%d')}.json"
        filepath = os.path.join(output_dir, filename)
        
        output = {
            'metadata': summary,
            'records': records,
        }
        
        with open(filepath, 'w') as f:
            json.dump(output, f, indent=2)
        
        print(f"Saved {len(records)} records to {filepath}")
        print(f"File size: {os.path.getsize(filepath)} bytes")
    
    load_task = PythonOperator(
        task_id='load_weather_data',
        python_callable=load_weather_data,
        provide_context=True,
    )

    # Task 4: Notification - Send a summary notification
    def notify_completion(**context):
        """Log a completion notification."""
        summary = context['task_instance'].xcom_pull(
            task_ids='transform_weather_data',
            key='summary'
        )
        
        print("=" * 50)
        print("PIPELINE COMPLETED SUCCESSFULLY")
        print(f"  Records processed: {summary.get('num_records', 0)}")
        print(f"  Avg Temperature: {summary.get('avg_temperature', 'N/A')}°C")
        print(f"  Processed at: {summary.get('processed_at', 'N/A')}")
        print("=" * 50)
    
    notify_task = PythonOperator(
        task_id='notify_completion',
        python_callable=notify_completion,
        provide_context=True,
    )

    # Define the pipeline structure using the bit-shift operator
    # This creates: fetch -> transform -> load -> notify
    fetch_task >> transform_task >> load_task >> notify_task

This pipeline demonstrates the classic Extract-Transform-Load (ETL) pattern. Let's walk through what each task does:

The dependency line fetch_task >> transform_task >> load_task >> notify_task creates a linear chain where each task runs only after its predecessor succeeds. Airflow handles this orchestration automatically — if the fetch fails, transform never runs; if transform fails after three retries, load is skipped and you get an email alert.

Place this file in your dags/ folder and restart the scheduler to pick up the new DAG:

docker compose restart scheduler

After about 30 seconds, refresh the Airflow UI and you'll see the weather_data_pipeline DAG. Toggle the pause/unpause switch to activate it, then click the "Trigger DAG" button to run it manually. Watch the tasks turn from white to green as they succeed, and click on any task to view its logs.

Building a More Advanced Pipeline

Real-world pipelines often involve conditional branching, parallel execution, and integration with external systems. Let's build a more sophisticated example that processes sales data with branching logic and database integration.

Create a second DAG at dags/sales_etl_pipeline.py:

# dags/sales_etl_pipeline.py
"""
Advanced ETL pipeline with branching, database operations,
and parallel task execution.
"""

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator, BranchPythonOperator
from airflow.operators.empty import EmptyOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook
import random
import json

default_args = {
    'owner': 'analytics_team',
    'depends_on_past': False,
    'retries': 2,
    'retry_delay': timedelta(minutes=3),
    'start_date': datetime(2024, 1, 1),
    'email_on_failure': True,
    'email': ['analytics-alerts@example.com'],
}

with DAG(
    dag_id='sales_etl_pipeline',
    default_args=default_args,
    description='Advanced ETL with branching and Postgres operations',
    schedule_interval='@daily',
    catchup=False,
    tags=['tutorial', 'sales', 'advanced'],
) as dag:

    # --- Configuration ---
    CREATE_TABLE_SQL = """
    CREATE TABLE IF NOT EXISTS sales_records (
        id SERIAL PRIMARY KEY,
        region VARCHAR(50),
        product_category VARCHAR(100),
        units_sold INTEGER,
        revenue DECIMAL(10, 2),
        sale_date DATE,
        processed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    );
    """

    INSERT_SALES_SQL = """
    INSERT INTO sales_records (region, product_category, units_sold, revenue, sale_date)
    VALUES
        ('North', 'Electronics', %(units_north_elec)s, %(rev_north_elec)s, CURRENT_DATE - 1),
        ('North', 'Clothing', %(units_north_cloth)s, %(rev_north_cloth)s, CURRENT_DATE - 1),
        ('South', 'Electronics', %(units_south_elec)s, %(rev_south_elec)s, CURRENT_DATE - 1),
        ('South', 'Clothing', %(units_south_cloth)s, %(rev_south_cloth)s, CURRENT_DATE - 1),
        ('East', 'Electronics', %(units_east_elec)s, %(rev_east_elec)s, CURRENT_DATE - 1),
        ('West', 'Clothing', %(units_west_cloth)s, %(rev_west_cloth)s, CURRENT_DATE - 1);
    """

    # --- Task: Ensure table exists ---
    create_table = PostgresOperator(
        task_id='create_sales_table',
        postgres_conn_id='postgres_default',
        sql=CREATE_TABLE_SQL,
    )

    # --- Task: Generate sales data ---
    def generate_sales_data(**context):
        """Simulate generating sales data from various sources."""
        import random
        
        data = {
            'params': {
                'units_north_elec': random.randint(50, 200),
                'rev_north_elec': round(random.uniform(5000, 20000), 2),
                'units_north_cloth': random.randint(30, 150),
                'rev_north_cloth': round(random.uniform(2000, 10000), 2),
                'units_south_elec': random.randint(40, 180),
                'rev_south_elec': round(random.uniform(4000, 18000), 2),
                'units_south_cloth': random.randint(20, 120),
                'rev_south_cloth': round(random.uniform(1500, 8000), 2),
                'units_east_elec': random.randint(60, 220),
                'rev_east_elec': round(random.uniform(6000, 22000), 2),
                'units_west_cloth': random.randint(25, 130),
                'rev_west_cloth': round(random.uniform(1800, 9000), 2),
            },
            'total_records': 6,
            'total_revenue': 0.0,
        }
        
        # Calculate total revenue
        revenue_fields = [k for k in data['params'] if k.startswith('rev_')]
        data['total_revenue'] = round(
            sum(data['params'][k] for k in revenue_fields), 2
        )
        
        context['task_instance'].xcom_push(key='sales_data', value=data)
        print(f"Generated sales data: {data['total_records']} records")
        print(f"Total revenue: ${data['total_revenue']:,.2f}")
        
        return data
    
    generate_data = PythonOperator(
        task_id='generate_sales_data',
        python_callable=generate_sales_data,
        provide_context=True,
    )

    # --- Task: Insert sales data into Postgres ---
    def insert_sales_to_db(**context):
        """Insert generated sales data into the sales_records table."""
        sales_data = context['task_instance'].xcom_pull(
            task_ids='generate_sales_data',
            key='sales_data'
        )
        
        if not sales_data:
            raise ValueError("No sales data available")
        
        postgres_hook = PostgresHook(postgres_conn_id='postgres_default')
        connection = postgres_hook.get_conn()
        cursor = connection.cursor()
        
        try:
            cursor.execute(
                INSERT_SALES_SQL,
                sales_data['params']
            )
            connection.commit()
            print(f"Inserted {sales_data['total_records']} records into sales_records")
        except Exception as e:
            connection.rollback()
            raise RuntimeError(f"Database insert failed: {e}")
        finally:
            cursor.close()
            connection.close()
    
    insert_data = PythonOperator(
        task_id='insert_sales_data',
        python_callable=insert_sales_to_db,
        provide_context=True,
    )

    # --- Task: Branch based on revenue threshold ---
    def decide_processing_path(**context):
        """Branch the pipeline based on total revenue."""
        sales_data = context['task_instance'].xcom_pull(
            task_ids='generate_sales_data',
            key='sales_data'
        )
        
        total_revenue = sales_data.get('total_revenue', 0) if sales_data else 0
        threshold = 50000
        
        if total_revenue >= threshold:
            print(f"Revenue ${total_revenue:,.2f} >= ${threshold:,} — "
                  "Running full analytics")
            return 'run_full_analytics'
        else:
            print(f"Revenue ${total_revenue:,.2f} < ${threshold:,} — "
                  "Running summary only")
            return 'run_summary_only'
    
    branch_task = BranchPythonOperator(
        task_id='branch_on_revenue',
        python_callable=decide_processing_path,
        provide_context=True,
    )

    # --- Branch A: Full analytics (high revenue) ---
    def run_full_analytics_fn(**context):
        """Perform comprehensive analytics on sales data."""
        postgres_hook = PostgresHook(postgres_conn_id='postgres_default')
        records = postgres_hook.get_records(
            sql="SELECT region, product_category, units_sold, revenue "
                "FROM sales_records "
                "WHERE sale_date = CURRENT_DATE - 1;"
        )
        
        # Detailed analysis
        regions = {}
        categories = {}
        for row in records:
            region, category, units, revenue = row
            regions[region] = regions.get(region, 0) + revenue
            categories[category] = categories.get(category, 0) + revenue
        
        print("=== FULL ANALYTICS REPORT ===")
        print(f"Total records analyzed: {len(records)}")
        print("Revenue by Region:")
        for region, rev in sorted(regions.items(), key=lambda x: x[1], reverse=True):
            print(f"  {region}: ${rev:,.2f}")
        print("Revenue by Category:")
        for cat, rev in sorted(categories.items(), key=lambda x: x[1], reverse=True):
            print(f"  {cat}: ${rev:,.2f}")
        
        top_region = max(regions, key=regions.get)
        print(f"Top Performing Region: {top_region} (${regions[top_region]:,.2f})")
    
    full_analytics = PythonOperator(
        task_id='run_full_analytics',
        python_callable=run_full_analytics_fn,
        provide_context=True,
    )

    # --- Branch B: Summary only (low revenue) ---
    def run_summary_only_fn(**context):
        """Generate a lightweight summary."""
        sales_data = context['task_instance'].xcom_pull(
            task_ids='generate_sales_data',
            key='sales_data'
        )
        
        print("=== SUMMARY REPORT ===")
        print(f"Total revenue: ${sales_data['total_revenue']:,.2f}")
        print(f"Records: {sales_data['total_records']}")
        print("Note: Revenue below threshold — skipping detailed analytics")
    
    summary_only = PythonOperator(
        task_id='run_summary_only',
        python_callable=run_summary_only_fn,
        provide_context=True,
    )

    # --- Join point after branching ---
    join_task = EmptyOperator(
        task_id='analytics_complete',
        trigger_rule='none_failed_min_one_success',
    )

    # --- Cleanup / archiving ---
    def archive_pipeline_run(**context):
        """Archive pipeline metadata."""
        execution_date = context['execution_date']
        dag_run_id = context['dag_run'].run_id
        
        metadata = {
            'dag_id': 'sales_etl_pipeline',
            'execution_date': str(execution_date),
            'run_id': dag_run_id,
            'completed_at': str(datetime.now()),
        }
        
        archive_path = '/opt/airflow/output/archive'
        os.makedirs(archive_path, exist_ok=True)
        
        filename = f"run_{execution_date.strftime('%Y%m%d_%H%M%S')}.json"
        filepath = os.path.join(archive_path, filename)
        
        with open(filepath, 'w') as f:
            json.dump(metadata, f, indent=2)
        
        print(f"Archived run metadata to {filepath}")
    
    archive_task = PythonOperator(
        task_id='archive_pipeline_run',
        python_callable=archive_pipeline_run,
        provide_context=True,
        trigger_rule='all_done',
    )

    # --- Define dependencies ---
    # Main flow: create_table >> generate_data >> insert_data >> branch
    create_table >> generate_data >> insert_data >> branch_task
    
    # Branching paths
    branch_task >> full_analytics >> join_task
    branch_task >> summary_only >> join_task
    
    # After join, archive
    join_task >> archive_task

This advanced pipeline introduces several powerful Airflow features:

To use the Postgres operators, you need to configure a connection. In the Airflow UI, go to Admin → Connections, create a new connection with conn_id = 'postgres_default', connection type Postgres, host postgres (matching the Docker service name), port 5432, database airflow, user airflow, and password airflow.

Best Practices for Airflow Data Pipelines

Building reliable, maintainable Airflow pipelines requires more than just getting the code to run. Here are the best practices that experienced Airflow developers follow:

1. Make Tasks Idempotent

Idempotency is the single most important principle in pipeline design. An idempotent task produces the same result no matter how many times it runs with the same inputs. Why does this matter? Air

🚀 Need a reliable AI agent for your project?

Deploy Hermes Agent in 10 minutes. Managed hosting, zero DevOps.

Get Started — $23.99/mo
← Back to all articles