Apache Airflow with Docker — Getting Started Tutorial

Environment: Apache Airflow running locally via Docker Compose UI: http://localhost:8080 Goal: Understand Airflow, set up your environment, and write your first DAG

Table of Contents

1. What Is Apache Airflow?
2. Key Concepts
3. Prerequisites
4. Project Structure
5. Setup & Installation
6. Common Commands
7. Your First DAG
8. A Real-World ETL DAG
9. Passing Data Between Tasks (XCom)
10. Using the Airflow UI
11. Best Practices
12. Troubleshooting
13. Next Steps

1. What Is Apache Airflow?

Apache Airflow is an open-source platform for authoring, scheduling, and monitoring workflows. Workflows are defined as Python code, which makes them version-controllable, testable, and highly flexible.

Airflow is ideal for:

• ETL pipelines (Extract → Transform → Load)
• Data engineering automation
• Scheduled jobs and reporting
• Orchestrating microservices or scripts

How it Works

Airflow workflows follow a simple mental model:

extract → transform → load → validate → notify

Each arrow is a task dependency. The whole pipeline is called a DAG.

2. Key Concepts

DAG Dependency Syntax

Airflow uses Python’s bitshift operators to define task order:

task1 >> task2          # task1 runs before task2
task1 >> [task2, task3] # task1 fans out to task2 and task3 in parallel
[task2, task3] >> task4 # task4 waits for both task2 and task3

3. Prerequisites

Before you begin, make sure you have the following installed:

• Docker Desktop (v20.10+) — https://docs.docker.com/get-docker/
• Docker Compose (v2.0+) — included with Docker Desktop
• At least 4 GB RAM allocated to Docker
• Python 3.8+ (optional, for local DAG linting)

Verify your Docker installation:

docker --version
docker compose version

4. Project Structure

Create a working directory for your project. Airflow with Docker Compose expects the following layout:

airflow-docker/
├── dags/               # ← Your Python DAG files go here
├── logs/               # Execution logs (auto-generated)
├── plugins/            # Custom operators, hooks, sensors
├── config/             # Optional airflow.cfg overrides
├── .env                # Environment variables (e.g., AIRFLOW_UID)
└── docker-compose.yaml # The Airflow stack definition

Tip: Only the dags/ folder matters day-to-day. Everything else is managed by Docker.

5. Setup & Installation

Step 1 — Create the project directory

mkdir airflow-docker && cd airflow-docker
mkdir -p dags logs plugins config

Step 2 — Download the official Docker Compose file

curl -LfO 'https://airflow.apache.org/docs/apache-airflow/stable/docker-compose.yaml'

This file defines all Airflow services: webserver, scheduler, triggerer, and PostgreSQL.

Step 3 — Set the Airflow UID

This prevents permission issues between Docker and your host filesystem:

echo -e "AIRFLOW_UID=$(id -u)" > .env

On Windows, use:

echo AIRFLOW_UID=50000 > .env

Step 4 — Initialize Airflow (first run only)

This creates the database schema and the default admin user:

docker compose up airflow-init

Wait for the output:

airflow-init_1  | Upgrades done
airflow-init_1  | Admin user airflow created
airflow-init_1 exited with code 0

Step 5 — Start all services

docker compose up

Or run in the background (detached mode):

docker compose up -d

Step 6 — Open the UI

Navigate to http://localhost:8080 in your browser.

Default credentials:

6. Common Commands

# Start all services
docker compose up

# Start in background (detached)
docker compose up -d

# Initialize Airflow (first run only)
docker compose up airflow-init

# Stop all services
docker compose down

# Stop and remove volumes (full reset)
docker compose down --volumes --remove-orphans

# View live logs
docker compose logs -f

# View logs for a specific service
docker compose logs -f airflow-scheduler

# Run a command inside the Airflow container
docker compose exec airflow-webserver airflow dags list

# Trigger a DAG from the CLI
docker compose exec airflow-webserver airflow dags trigger my_dag_id

7. Your First DAG

Create a new file in the dags/ folder:

touch dags/hello_world.py

Paste the following:

# dags/hello_world.py

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator


# --- Default arguments applied to all tasks ---
default_args = {
    "owner": "airflow",
    "depends_on_past": False,
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
}


# --- DAG definition ---
with DAG(
    dag_id="hello_world",
    description="A simple Hello World DAG",
    default_args=default_args,
    start_date=datetime(2024, 1, 1),
    schedule_interval="@daily",   # Run once per day
    catchup=False,                # Don't backfill missed runs
    tags=["tutorial"],
) as dag:

    # Task 1: Run a bash command
    say_hello = BashOperator(
        task_id="say_hello",
        bash_command="echo 'Hello from Airflow!'",
    )

    # Task 2: Run a Python function
    def print_date():
        print(f"Today is: {datetime.now().strftime('%Y-%m-%d')}")

    print_today = PythonOperator(
        task_id="print_today",
        python_callable=print_date,
    )

    # Task order: say_hello runs first, then print_today
    say_hello >> print_today

The Airflow Scheduler automatically detects new files in dags/ — no restart required.

Within 30–60 seconds, hello_world will appear in the UI at http://localhost:8080.

Anatomy of a DAG

with DAG(
    dag_id="...",         # Unique identifier — shown in the UI
    start_date=...,       # When Airflow begins scheduling this DAG
    schedule_interval=...,# Cron expression or preset (@daily, @hourly, None)
    catchup=False,        # Whether to run missed historical intervals
    tags=[...],           # Labels for filtering in the UI
) as dag:
    ...

Common Schedule Intervals

8. A Real-World ETL DAG

This example simulates a typical data pipeline:

extract_data → transform_data → load_data → validate_data → notify_success

# dags/etl_pipeline.py

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator


default_args = {
    "owner": "data-team",
    "retries": 2,
    "retry_delay": timedelta(minutes=3),
    "email_on_failure": False,
}


def extract(**context):
    """Simulate extracting data from a source system."""
    print("Extracting data from source...")
    records = [{"id": 1, "value": 100}, {"id": 2, "value": 200}]
    # Push data to XCom for the next task
    context["ti"].xcom_push(key="raw_records", value=records)
    print(f"Extracted {len(records)} records.")


def transform(**context):
    """Simulate transforming the extracted data."""
    ti = context["ti"]
    records = ti.xcom_pull(task_ids="extract_data", key="raw_records")
    print(f"Transforming {len(records)} records...")
    transformed = [{"id": r["id"], "value": r["value"] * 1.1} for r in records]
    ti.xcom_push(key="transformed_records", value=transformed)
    print("Transformation complete.")


def load(**context):
    """Simulate loading data into a destination."""
    ti = context["ti"]
    records = ti.xcom_pull(task_ids="transform_data", key="transformed_records")
    print(f"Loading {len(records)} records to destination...")
    # In production: write to a database, S3, BigQuery, etc.
    print("Load complete.")


def validate(**context):
    """Run data quality checks."""
    print("Running validation checks...")
    # In production: assert row counts, check nulls, etc.
    print("All checks passed.")


with DAG(
    dag_id="etl_pipeline",
    description="Extract → Transform → Load → Validate",
    default_args=default_args,
    start_date=datetime(2024, 1, 1),
    schedule_interval="0 6 * * *",  # Daily at 6am
    catchup=False,
    tags=["etl", "tutorial"],
) as dag:

    extract_data = PythonOperator(
        task_id="extract_data",
        python_callable=extract,
    )

    transform_data = PythonOperator(
        task_id="transform_data",
        python_callable=transform,
    )

    load_data = PythonOperator(
        task_id="load_data",
        python_callable=load,
    )

    validate_data = PythonOperator(
        task_id="validate_data",
        python_callable=validate,
    )

    notify_success = BashOperator(
        task_id="notify_success",
        bash_command="echo 'Pipeline completed successfully at $(date)'",
    )

    # Define the pipeline order
    extract_data >> transform_data >> load_data >> validate_data >> notify_success

9. Passing Data Between Tasks (XCom)

XCom (Cross-Communication) lets tasks share small pieces of data.

# Pushing a value
def my_producer(**context):
    context["ti"].xcom_push(key="result", value=42)

# Pulling a value
def my_consumer(**context):
    value = context["ti"].xcom_pull(task_ids="producer_task", key="result")
    print(f"Got: {value}")

Important: XCom is designed for small data (strings, numbers, small dicts/lists). For large datasets, write to a file or database and pass the path or identifier instead.

10. Using the Airflow UI

Once running at http://localhost:8080, here’s what to know:

DAGs List View

The home screen shows all your DAGs, their schedule, last run status, and toggle to enable/disable them.

Graph View

Click a DAG → Graph tab to see the visual task dependency diagram. This is the best view for understanding pipeline structure.

Grid View

Shows a history of all DAG runs as a grid. Click any cell to drill into a specific run and see task logs.

Triggering a DAG Manually

1. Go to the DAGs list
2. Click the ▶️ Trigger DAG button on the right side of the row
3. Optionally pass configuration JSON

Viewing Task Logs

1. Click on a DAG run
2. Click on a task node
3. Click Log to see stdout/stderr output

11. Best Practices

DAG Design

• Keep DAGs idempotent — running the same DAG twice should produce the same result
• Set catchup=False unless you intentionally want to backfill
• Use tags to organize DAGs in large projects
• Set max_active_runs to prevent overlapping runs on slow pipelines

Task Design

• Keep tasks small and focused on a single responsibility
• Use retries and retry_delay on tasks that call external services
• Avoid heavy computation inside DAG-level code (it runs frequently during parsing)

Code Organization

dags/
├── etl/
│   ├── extract.py
│   └── pipeline.py
├── reporting/
│   └── daily_report.py
└── utils/
    └── helpers.py       # Shared helper functions

Security

• Never hardcode credentials in DAG files
• Use Airflow Connections (Admin → Connections in the UI) to store secrets
• Reference them in code with BaseHook.get_connection("my_conn_id")

12. Troubleshooting

DAG doesn’t appear in the UI

• Check for Python syntax errors: python dags/my_dag.py
• Wait 30–60 seconds for the scheduler to pick it up
• Check scheduler logs: docker compose logs -f airflow-scheduler

DAG shows “Import Error” in the UI

• Click the DAG name to see the error message
• Usually caused by a missing import or syntax error in the DAG file

Port 8080 already in use

Edit docker-compose.yaml and change the webserver port mapping:

ports:
  - "8081:8080"   # Access UI at localhost:8081

Tasks stuck in “queued” state

The scheduler may not have enough workers. Restart services:

docker compose down && docker compose up

Full reset (wipe all data and start fresh)

docker compose down --volumes --remove-orphans
docker compose up airflow-init
docker compose up

13. Next Steps

Once you’re comfortable with the basics, explore these areas:

Useful Resources

• Official Airflow Documentation
• Airflow Tutorials
• Astronomer Registry — community DAGs and providers
• Airflow GitHub

Claude Context Block: This project runs Airflow locally via Docker Compose. DAGs live in dags/, the UI is at localhost:8080, and the stack includes a PostgreSQL metadata database. When asking for help, mention the operator type, error message, or DAG structure for the most targeted assistance.

• Airflow
• Docker
• Data-Engineering
• Orchestration
• Etl
• Python
• Devops