Build an ETL Pipeline with Airflow

Build a scheduled ETL pipeline with Apache Airflow running locally via Docker.

Introduction

30 Second Summary

Every team has that collection of Python scripts running on a schedule. Maybe yours runs nightly reports or processes daily orders. But when something breaks at 2 AM, nobody knows until someone manually checks the next morning.

In this project, you will build an automated ETL pipeline using Apache Airflow. You'll extract sales data, transform it into useful metrics, load the results to a file, and monitor every step through a visual dashboard.

What You'll Build

A local Airflow instance running a mini ETL pipeline that extracts sales order data, computes totals and averages, and writes the results to a JSON file on an automatic schedule.

By the end of this project, you'll have:

  • A running Airflow environment with webserver, scheduler, and worker containers powered by Docker Compose.
  • A Python DAG file defining an Extract-Transform-Load pipeline that passes data between tasks.
  • A scheduled workflow with automatic retries on failure and full visibility through the Airflow Web UI.
  • Secret Mission: Add a parallel validation branch that checks data quality before loading.

Are there any prerequisites?

You should be familiar with basic Python (functions, dictionaries, and imports). You'll also need Docker Desktop installed on Windows. No prior Airflow experience is needed.

Before We Start

Before We Start...

Set Up Airflow with Docker Compose

Before you can build pipelines, you need a working Apache Airflow environment. Installing Airflow directly on Windows is tricky because of its many dependencies (a web server, a scheduler, a task worker, a database, and a message broker).

Docker Compose solves this by bundling all of those services into containers that run together with a single command. In this step, you will download the official Airflow Docker Compose configuration, start all the services, and verify the Web UI is running.

In this step, get ready to:
  • Create the Airflow project folder and download the official Docker Compose configuration.
  • Initialize the Airflow database and start all services.
  • Log in to the Airflow Web UI and confirm the dashboard is live.
Download the official docker-compose.yaml and create the project structure

Airflow publishes an official docker-compose.yaml file that defines all the services you need (web server, scheduler, worker, PostgreSQL database, and Redis message broker). You will download this file into a new project folder, then create the local directories Airflow expects for storing DAG files, logs, plugins, and configuration.

  • Open PowerShell by pressing the Windows key, typing PowerShell, and pressing Enter.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir airflow-project
cd airflow-project
  • Download the official Airflow 3.2.2 Docker Compose file by running:
curl -LfO "https://airflow.apache.org/docs/apache-airflow/3.2.2/docker-compose.yaml"

What does this command do?

curl -LfO downloads a file from a URL and saves it with its original filename. The -L flag follows redirects, -f fails silently on HTTP errors, and -O keeps the remote filename (docker-compose.yaml).

  • Create the four subdirectories Airflow requires by running:
mkdir dags, logs, plugins, config

What are these folders for?

  • dags/ holds your pipeline definition files (Python scripts).
  • logs/ stores task execution logs so you can debug failures.
  • plugins/ is for custom Airflow plugins (not needed today, but required by the config).
  • config/ holds optional Airflow configuration overrides.
  • Confirm your project structure by running:
ls

You should see docker-compose.yaml and the four folders (config, dags, logs, plugins) listed in the output.

Don't see docker-compose.yaml?

If the curl command printed an error, make sure Docker Desktop is running and your internet connection is active. Re-run the curl command above.

help me troubleshoot the curl download for docker-compose.yaml

  • Create the .env file that sets the Airflow user ID by running:
Set-Content .env "AIRFLOW_UID=50000"

Why do I need this .env file?

On Windows, Docker does not automatically map your host user ID into the container. Setting AIRFLOW_UID=50000 tells the containers which user ID to run as, preventing a permissions warning when you start the services.

✔️ Awesome, I've got everything!

Great. Double check you have all 6 items when you run ls (the .env file, docker-compose.yaml, and the four folders).

ⓧ I'd like to double check the full code

Your .env file should contain exactly this:

AIRFLOW_UID=50000

And running ls in the airflow-project folder should show:

ls
# Expected output:
# .env
# config
# dags
# docker-compose.yaml
# logs
# plugins
Initialize the Airflow database and start all services

Airflow stores metadata (which DAGs exist, their run history, task statuses) in a PostgreSQL database. Before the services can start, you need to initialize that database and create the default admin user.

  • Initialize the Airflow database and create the admin user by running:
docker compose up airflow-init

What does this command do?

docker compose up airflow-init runs a one-time setup container that creates the database tables Airflow needs and a default admin account (username airflow, password airflow).

This will pull several Docker images on first run, so it may take a few minutes depending on your internet speed.

Wait for the command to finish. You will know it is done when you see a line containing exited with code 0 in the output.

Seeing an error about insufficient memory?

Airflow's containers need at least 4 GB of memory. Open Docker Desktop, go to Settings then Resources, and increase the memory allocation to at least 4 GB.

help me fix Docker memory issues for Airflow

  • Start all Airflow services in the background by running:
docker compose up -d

What does the -d flag do?

The -d flag stands for "detached" mode. It starts the containers in the background so you get your terminal back. Without it, all container logs would stream into your terminal window and you would have to open a new one to keep working.

  • Wait about 30-60 seconds for all services to become healthy, then verify the containers are running:
docker compose ps

You should see several containers listed with a status that includes Up or healthy. These include the webserver, scheduler, worker, PostgreSQL database, and Redis message broker.

Containers showing as unhealthy or restarting?

Give it another minute. The webserver container can take up to 60 seconds to pass its health check. Run docker compose ps again after waiting.

If containers keep restarting, check that Docker Desktop has at least 4 GB of memory allocated.

help me debug my Airflow Docker containers

Verify the Airflow Web UI is running

The Airflow Web UI is a dashboard where you (and your team) can monitor all pipelines, check run history, and drill into task logs. If the containers started successfully, the UI is already live.

  • Open your browser and navigate to http://localhost:8080.
  • Enter airflow in the Username field.
  • Enter airflow in the Password field.
  • Click Sign In.

You should see the Airflow dashboard with a list of example DAGs (Directed Acyclic Graphs). These are pre-loaded sample pipelines that ship with the official Docker image.

What is a DAG?

A DAG is a collection of tasks with defined dependencies. "Directed" means tasks flow in one direction (A runs before B). "Acyclic" means no task can loop back to a previous one.

Every pipeline you build in Airflow is defined as a DAG. The dashboard you are looking at right now is where your team monitors them all.

Page not loading at localhost:8080?

The webserver container may still be starting up. Wait another 30 seconds and refresh the page.

If it still does not load, go back to PowerShell and run docker compose ps to check that the webserver container status shows healthy.

help me get the Airflow Web UI working

Your Airflow environment is up and running. Next up, you will write your first ETL pipeline as a Python DAG and watch it execute through this dashboard.

Write Your First ETL DAG

Your Airflow environment is up and running. Now it is time to write the actual pipeline that Airflow will orchestrate.

In this step, you will create a Python file that defines a DAG (Directed Acyclic Graph). A DAG describes your workflow as a series of tasks with clear dependencies between them. Your DAG will follow the ETL pattern: Extract raw data, Transform it into something useful, and Load the results to a file.

In this step, get ready to:
  • Create a DAG file that Airflow automatically detects.
  • Define extract and transform tasks that pass data using XCom.
  • Wire up task dependencies so they run in the correct order.
Create the DAG file with imports and configuration

Airflow automatically scans the dags/ folder for Python files containing DAG definitions. You will create your pipeline file there and set up the DAG context.

  • Create a new file called sales_etl_pipeline.py inside the dags/ folder of your airflow-project directory.
  • Add the following imports and DAG configuration to the file:
import json
import os
import pendulum

from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import DAG


with DAG(
    "sales_etl_pipeline",
    description="A simple ETL pipeline that extracts, transforms, and loads sales data",
    schedule=None,
    start_date=pendulum.datetime(2024, 1, 1, tz="UTC"),
    catchup=False,
    tags=["etl", "sales"],
) as dag:
    pass  # Tasks will go here

What does this code do?

The imports pull in the libraries your pipeline needs. json handles data serialization. pendulum provides timezone-aware dates. PythonOperator lets you run Python functions as Airflow tasks.

The with DAG(...) as dag: block is a context manager that groups all your tasks under one pipeline. schedule=None means the DAG only runs when you trigger it manually. catchup=False prevents Airflow from running the DAG for every missed interval since the start_date.

Why these import paths?

Airflow 3.2.2 uses different import paths than older versions you might find in tutorials online. The DAG object now lives in airflow.sdk and PythonOperator lives in airflow.providers.standard.operators.python. Using the old paths will cause import errors.

  • Save dags/sales_etl_pipeline.py.

File not appearing in the Airflow UI?

Airflow scans the dags/ folder periodically. It may take up to a minute for a new file to appear. Make sure your file is saved directly inside dags/ (not in a subfolder) and that the filename ends with .py.

If it still does not appear, check the file for Python syntax errors. Airflow silently skips files it cannot parse.

help me debug why my DAG file is not showing up in the Airflow UI

Define the extract and transform tasks using XCom

Each task in your DAG is a Python function. Tasks pass data between each other using XCom (cross-communication). One task pushes data, and the next task pulls it.

  • In dags/sales_etl_pipeline.py, remove the pass placeholder inside the with DAG(...) block and replace it with the extract function:
    def extract(**kwargs):
        """Extract sales order data from source."""
        ti = kwargs["ti"]
        # Simulated sales order data (in a real pipeline, this would come from a database or API)
        order_data = {
            "1001": 301.27,
            "1002": 433.21,
            "1003": 502.22,
            "1004": 115.90,
            "1005": 678.45,
        }
        # Push the data to XCom so downstream tasks can access it
        ti.xcom_push("order_data", json.dumps(order_data))
        print(f"Extracted {len(order_data)} orders")

What does this code do?

The extract function simulates pulling sales orders from a source system. In a real pipeline, this would query a database or call an API.

The key line is ti.xcom_push("order_data", json.dumps(order_data)). This stores the data in Airflow's XCom system so that later tasks can retrieve it. The ti object (task instance) is automatically passed to your function by Airflow through **kwargs.

  • Add the transform function directly below the extract function (still inside the with DAG(...) block):
    def transform(**kwargs):
        """Transform data: compute total and average order values."""
        ti = kwargs["ti"]
        # Pull the data that the extract task pushed to XCom
        order_data_str = ti.xcom_pull(task_ids="extract", key="order_data")
        order_data = json.loads(order_data_str)

        # Calculate summary statistics
        total_value = sum(order_data.values())
        avg_value = total_value / len(order_data)

        results = {
            "total_order_value": round(total_value, 2),
            "average_order_value": round(avg_value, 2),
            "order_count": len(order_data),
        }
        # Push transformed results for the load task
        ti.xcom_push("transformed_results", json.dumps(results))
        print(f"Transformed: total={results['total_order_value']}, avg={results['average_order_value']}")

How does XCom pull work?

The ti.xcom_pull(task_ids="extract", key="order_data") call reaches back to the extract task's output and retrieves the data it pushed. You specify which task to pull from (task_ids) and which key to look for.

XCom is designed for small amounts of metadata between tasks. For large datasets in production, you would write to external storage instead.

  • Save dags/sales_etl_pipeline.py.

Seeing indentation errors?

Both functions must be indented one level inside the with DAG(...) block (4 spaces). Python is strict about consistent indentation. Make sure you have not mixed tabs and spaces.

help me fix indentation errors in my Airflow DAG file

Define the load task and wire up dependencies

The final piece is the load task that writes results to a file, plus the operator definitions and dependency chain that tell Airflow the execution order.

  • Add the load function below the transform function:
    def load(**kwargs):
        """Load transformed results to a JSON file."""
        ti = kwargs["ti"]
        # Pull the transformed results from XCom
        results_str = ti.xcom_pull(task_ids="transform", key="transformed_results")
        results = json.loads(results_str)

        # Create the output directory and write results
        output_dir = "/opt/airflow/dags/output"
        os.makedirs(output_dir, exist_ok=True)
        output_path = os.path.join(output_dir, "results.json")

        with open(output_path, "w") as f:
            json.dump(results, f, indent=2)

        print(f"Loaded results to {output_path}: {results}")

What does this code do?

The load function pulls the transformed results from XCom and writes them to /opt/airflow/dags/output/results.json. This path is inside the Airflow container, mapped to your local dags/ folder through Docker volumes.

The os.makedirs(output_dir, exist_ok=True) line creates the output folder if it does not already exist. This prevents the task from crashing on the first run.

  • Add the PythonOperator task definitions and the dependency chain below the load function:
    # Create task operators that wrap each Python function
    extract_task = PythonOperator(
        task_id="extract",
        python_callable=extract,
    )

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

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

    # Set task dependencies: extract runs first, then transform, then load
    extract_task >> transform_task >> load_task

What does the >> operator do?

The >> operator sets downstream dependencies. extract_task >> transform_task >> load_task tells Airflow: run extract first. Only after extract succeeds, run transform. Only after transform succeeds, run load.

If extract fails, Airflow will not attempt to run transform or load. This prevents corrupted data from flowing downstream.

  • Save dags/sales_etl_pipeline.py.
  • Return to the Airflow Web UI at http://localhost:8080. Within about a minute, you should see sales_etl_pipeline appear in the DAG list.

DAG still not appearing after a minute?

Check for import errors by looking at the top of the DAG list page in the UI for a red banner. Import errors usually mean a typo in one of the import paths or a syntax error in your Python file.

Make sure you are using from airflow.sdk import DAG and from airflow.providers.standard.operators.python import PythonOperator. These are the correct import paths for Airflow 3.2.2.

help me figure out why my sales_etl_pipeline DAG has import errors in Airflow

✔️ Awesome, I've got everything!

Great. Double check you have saved your file.

ⓧ I'd like to double check the full code

import json
import os
import pendulum

from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import DAG


with DAG(
    "sales_etl_pipeline",
    description="A simple ETL pipeline that extracts, transforms, and loads sales data",
    schedule=None,
    start_date=pendulum.datetime(2024, 1, 1, tz="UTC"),
    catchup=False,
    tags=["etl", "sales"],
) as dag:

    def extract(**kwargs):
        """Extract sales order data from source."""
        ti = kwargs["ti"]
        order_data = {
            "1001": 301.27,
            "1002": 433.21,
            "1003": 502.22,
            "1004": 115.90,
            "1005": 678.45,
        }
        ti.xcom_push("order_data", json.dumps(order_data))
        print(f"Extracted {len(order_data)} orders")

    def transform(**kwargs):
        """Transform data: compute total and average order values."""
        ti = kwargs["ti"]
        order_data_str = ti.xcom_pull(task_ids="extract", key="order_data")
        order_data = json.loads(order_data_str)

        total_value = sum(order_data.values())
        avg_value = total_value / len(order_data)

        results = {
            "total_order_value": round(total_value, 2),
            "average_order_value": round(avg_value, 2),
            "order_count": len(order_data),
        }
        ti.xcom_push("transformed_results", json.dumps(results))
        print(f"Transformed: total={results['total_order_value']}, avg={results['average_order_value']}")

    def load(**kwargs):
        """Load transformed results to a JSON file."""
        ti = kwargs["ti"]
        results_str = ti.xcom_pull(task_ids="transform", key="transformed_results")
        results = json.loads(results_str)

        output_dir = "/opt/airflow/dags/output"
        os.makedirs(output_dir, exist_ok=True)
        output_path = os.path.join(output_dir, "results.json")

        with open(output_path, "w") as f:
            json.dump(results, f, indent=2)

        print(f"Loaded results to {output_path}: {results}")

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

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

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

    extract_task >> transform_task >> load_task

Your ETL pipeline is defined and Airflow can see it. Next up, you will trigger the DAG and watch each task execute in real time through the Graph view.

Trigger and Monitor Your Pipeline

Your DAG file is written and sitting in the dags/ folder, but it hasn't actually run yet. Right now it's just a set of instructions waiting to be executed.

In this step, you'll trigger your pipeline manually through the Airflow Web UI and watch each task execute in sequence. You'll also inspect the data flowing between tasks to confirm everything connects correctly.

In this step, get ready to:
  • Find and trigger your DAG from the Airflow dashboard.
  • Watch your ETL pipeline execute in the Graph view.
  • Inspect XCom values and task logs to verify data flow.
Find and trigger your DAG in the Web UI

Airflow scans the dags/ folder periodically for new Python files. Your sales_etl_pipeline DAG should already be visible in the dashboard. If it hasn't appeared yet, wait about a minute and refresh the page.

  • Return to the Airflow dashboard at http://localhost:8080.
  • Find sales_etl_pipeline in the DAG list.
  • Click the toggle switch next to the DAG name to unpause it.
  • Click the play button (▶) on the right side of the row to trigger a manual run.

What does unpausing do?

New DAGs start in a paused state by default. While paused, Airflow won't execute the DAG on its schedule. Unpausing enables the scheduler to pick it up automatically.

Even after unpausing, clicking the play button lets you trigger an immediate run without waiting for the next scheduled interval.

Don't see your DAG in the list?

Airflow may take up to a minute to detect new DAG files. Refresh the page and wait.

If it still doesn't appear, check that your file is saved as dags/sales_etl_pipeline.py inside the airflow-project folder. Airflow only scans the dags/ directory that's mounted into the container.

Help me figure out why my DAG isn't showing up in the Airflow UI

Explore the DAG run in the Graph view

The Graph view is where Airflow's value becomes obvious. Instead of guessing whether a script ran successfully, you can see every task's status at a glance.

  • Click on the DAG name sales_etl_pipeline to open the DAG detail page.
  • Click the Graph tab to see your three tasks displayed as a visual pipeline.
  • Watch as each task turns green in sequence: extract first, then transform, then load.

Once all three tasks show green, your ETL pipeline has completed successfully. The output file is now written inside the container.

Why is this better than Windows Task Scheduler?

Windows Task Scheduler tells you a script was triggered. Airflow tells you which step in the script succeeded, which failed, what data it processed, and lets your entire team see that status from their own browser.

Seeing a red or orange task?

A red task means it failed. An orange task means it's retrying. Click the coloured task box in the Graph view, then click Logs to see the error message. The most common cause is a syntax error in sales_etl_pipeline.py.

Help me debug my failed Airflow task

View XCom values and task logs

Your tasks pass data between each other using XCom (cross-communication). You can inspect exactly what data each task pushed and pulled directly from the UI.

  • In the Graph view, click on the extract task box.
  • Click XCom to see the data this task pushed.
  • You should see a key called order_data containing the JSON string of your five orders.
  • Close the panel and click the transform task box.
  • Click XCom again. You should see a key called transformed_results with the computed totals and averages.

What should I see in XCom?

The extract task's XCom shows the raw order data as a JSON string with order IDs and dollar amounts.

The transform task's XCom shows the computed results: a total order value of 2031.05, an average of 406.21, and an order count of 5.

Now check the logs to see the printed output from each function.

  • Click on the extract task box again.
  • Click Logs to open the task log output.
  • Scroll through the log. You should see the line Extracted 5 orders printed by your extract function.
  • Repeat for the transform and load tasks to confirm they ran correctly.

What do the logs tell me?

Every print() statement in your Python functions appears here. This is how you debug pipelines in production. Instead of checking script output files or guessing what happened, your team reads the logs directly from the dashboard.

Can't find the XCom or Logs tab?

Make sure you clicked on the task box in the Graph view first. A panel should appear showing details for that specific task instance. The XCom and Logs options are inside that panel.

Help me find the XCom and Logs tabs in the Airflow UI

Your pipeline runs, passes data between tasks, and produces a verified output. Next up, you'll add scheduling and retry logic so this pipeline runs automatically and recovers from failures on its own.

Add Scheduling and Retry Logic

You've triggered your DAG manually and watched all three tasks run to completion. In a real team environment, you need pipelines that run themselves on a schedule and recover gracefully when something goes wrong.

In this step, you'll explore the scheduling and retry configuration already in your DAG file. Then you'll intentionally break a task to watch Airflow's automatic retry logic in action. This is the exact behavior that replaces manually checking whether your scripts ran overnight.

In this step, get ready to:
  • Understand the cron schedule and retry configuration in your DAG.
  • Simulate a task failure and observe automatic retries in the Web UI.
  • Fix the failure and verify your pipeline recovers on its next scheduled run.
Understand the cron schedule and retry configuration

Your sales_etl_pipeline.py file already includes scheduling and retry logic from when you wrote it. Let's break down what those lines do.

  • In sales_etl_pipeline.py, find the schedule parameter inside the DAG() constructor:
    schedule="*/5 * * * *",

What does this cron expression mean?

The string */5 * * * * is a cron expression that means "every 5 minutes." The five positions represent minute, hour, day of month, month, and day of week.

For testing, every 5 minutes lets you quickly see scheduled runs appear. In production, you'd use something like 0 0 * * * for daily at midnight.

  • Now find the default_args dictionary near the top of the DAG definition:
    default_args={
        "retries": 2,
        "retry_delay": pendulum.duration(seconds=30),
    },

What does default_args do?

The default_args dictionary applies settings to every task in the DAG. Here you've configured two behaviors:

  • retries: 2 tells Airflow to automatically retry a failed task up to 2 times before marking it as permanently failed.
  • retry_delay: pendulum.duration(seconds=30) waits 30 seconds between each retry attempt, giving transient issues (like a network timeout) time to resolve.

Why pendulum instead of timedelta?

Airflow already depends on the pendulum library for date and time handling throughout the framework. Using pendulum.duration() keeps your code consistent with Airflow's internal patterns.

  • Also note the catchup=False parameter in your DAG definition.

What is catchup?

When catchup=False, Airflow skips any scheduled runs that would have occurred between the start_date and now. Without this, Airflow would try to backfill every missed 5-minute interval since January 2024.

  • Return to the Airflow Web UI at http://localhost:8080.
  • Check that your sales_etl_pipeline DAG is unpaused (the toggle switch next to the name should be blue/on).
  • Wait up to 5 minutes and you should see a new DAG run appear automatically in the list. The Run Type column will show "scheduled" instead of "manual."

Not seeing scheduled runs?

Make sure the DAG toggle is switched on (unpaused). A paused DAG won't execute scheduled runs even if the cron expression is valid.

Also check that your file is saved. Airflow rescans the dags/ folder periodically, so changes may take up to a minute to take effect.

Why isn't my DAG running on schedule in Airflow?

Simulate a failure and observe retry behavior

Retries are your safety net for transient failures like network timeouts or temporary database locks. Let's trigger a failure on purpose so you can see exactly how Airflow handles it.

  • In dags/sales_etl_pipeline.py, find this line inside the transform function:
        avg_value = total_value / len(order_data)
  • Replace len(order_data) with 0 so the line now looks like this:
        avg_value = total_value / 0  # Intentional divide-by-zero to test retries
  • Save sales_etl_pipeline.py.
  • Return to the Airflow Web UI at http://localhost:8080.
  • Click the play button (▶) next to sales_etl_pipeline to trigger a manual run.
  • Click on the DAG name to open it, then navigate to the Graph view.

Watch the task colors change. The extract task turns green (success), then the transform task turns yellow (up_for_retry) after it fails. After 30 seconds it retries, fails again, retries once more, and finally turns red (failed).

  • Click on the red transform task, then click Logs to see the failure message and retry attempts.

What should I see in the logs?

You'll see a ZeroDivisionError traceback for each attempt. The log also shows which retry attempt the task is on (attempt 1, 2, then 3 for the final failure). Notice that the load task never ran because its upstream dependency (transform) failed.

Transform task not turning red?

With 2 retries and a 30-second delay between each, it takes about 60-90 seconds for the task to exhaust all attempts and turn red. Be patient.

If the task still shows green (success), Airflow may not have detected your file change yet. Wait a minute, then trigger another run.

Help me debug why my intentional failure isn't showing up in Airflow.

Fix the failure and verify recovery

Now that you've seen how Airflow handles failures, let's restore the correct code and confirm your pipeline recovers on its next run.

  • In dags/sales_etl_pipeline.py, find the broken line inside the transform function:
        avg_value = total_value / 0  # Intentional divide-by-zero to test retries
  • Change it back to the original:
        avg_value = total_value / len(order_data)
  • Save sales_etl_pipeline.py.
  • Return to the Airflow Web UI and wait for the next scheduled run (up to 5 minutes), or click the play button (▶) to trigger a manual run immediately.
  • Watch the DAG run complete. All three tasks should turn green.

Click on the DAG name and look at the run history. You can see the failed run (red) alongside the successful recovery (green). This is exactly what replaces manual verification. Your team can see at a glance which pipelines are healthy and which need attention.

Why does this replace Windows Task Scheduler?

With Task Scheduler, a failed script leaves no visible trace unless someone manually checks the log file. Airflow gives you a visual dashboard where failures are immediately obvious, retries happen automatically, and your whole team sees the same status page.

Still seeing failures after fixing the code?

Airflow caches DAG files. After saving your fix, wait at least a minute for the scheduler to pick up the change before triggering a new run.

Double check that the line reads total_value / len(order_data) and not total_value / 0. A missed edit will keep causing the same failure.

Help me figure out why my DAG is still failing after I fixed the code.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file with the original len(order_data) code restored.

ⓧ I'd like to double check the full code

Here's the complete dags/sales_etl_pipeline.py file for comparison:

import json
import os
import pendulum

from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import DAG


with DAG(
    "sales_etl_pipeline",
    default_args={
        "retries": 2,
        "retry_delay": pendulum.duration(seconds=30),
    },
    description="A simple ETL pipeline that extracts, transforms, and loads sales data",
    schedule="*/5 * * * *",
    start_date=pendulum.datetime(2024, 1, 1, tz="UTC"),
    catchup=False,
    tags=["etl", "sales"],
) as dag:

    def extract(**kwargs):
        """Extract sales order data from source."""
        ti = kwargs["ti"]
        order_data = {
            "1001": 301.27,
            "1002": 433.21,
            "1003": 502.22,
            "1004": 115.90,
            "1005": 678.45,
        }
        ti.xcom_push("order_data", json.dumps(order_data))
        print(f"Extracted {len(order_data)} orders")

    def transform(**kwargs):
        """Transform data: compute total and average order values."""
        ti = kwargs["ti"]
        order_data_str = ti.xcom_pull(task_ids="extract", key="order_data")
        order_data = json.loads(order_data_str)

        total_value = sum(order_data.values())
        avg_value = total_value / len(order_data)

        results = {
            "total_order_value": round(total_value, 2),
            "average_order_value": round(avg_value, 2),
            "order_count": len(order_data),
        }
        ti.xcom_push("transformed_results", json.dumps(results))
        print(f"Transformed: total={results['total_order_value']}, avg={results['average_order_value']}")

    def load(**kwargs):
        """Load transformed results to a JSON file."""
        ti = kwargs["ti"]
        results_str = ti.xcom_pull(task_ids="transform", key="transformed_results")
        results = json.loads(results_str)

        output_dir = "/opt/airflow/dags/output"
        os.makedirs(output_dir, exist_ok=True)
        output_path = os.path.join(output_dir, "results.json")

        with open(output_path, "w") as f:
            json.dump(results, f, indent=2)

        print(f"Loaded results to {output_path}: {results}")

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

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

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

    extract_task >> transform_task >> load_task

Secret mission

Add a Parallel Data Validation Branch

Your ETL pipeline runs tasks in a straight line: extract, then transform, then load. But what if bad data sneaks through? In production, teams add validation gates that run alongside the transform to catch problems before they reach the output. Your mission: add a parallel validation branch that blocks the load if data quality checks fail.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally via Docker, so there are no cloud costs. The Airflow containers use approximately 4 GB of RAM while running.

Resources you used:

  • Docker containers (Airflow webserver, scheduler, worker, PostgreSQL database, Redis).
  • Local airflow-project folder on your Desktop (contains DAGs, logs, plugins, config, and docker-compose.yaml).

Keep everything running

No action needed. Choose this if you're still actively building or experimenting with your Airflow pipeline.

  • Your Airflow Web UI stays accessible at http://localhost:8080.
  • The scheduled DAG continues running every 5 minutes.
  • Docker Desktop continues using approximately 4 GB of RAM for the Airflow containers.

Pause - I'll come back to this later

Stop the containers without losing data. Your DAG files, logs, and database state are preserved so you can resume later.

  • Open PowerShell and navigate to your airflow-project folder on the Desktop.
  • Stop all containers by running:
cd ~/Desktop/airflow-project
docker compose stop
  • To resume later, run docker compose up -d from the same folder.

Delete - I don't want to use this again

Remove all containers, volumes, and project files entirely. This frees up all disk space and RAM used by the project.

  • Open PowerShell and navigate to your airflow-project folder on the Desktop.
  • Remove all Docker containers, networks, and volumes by running:
cd ~/Desktop/airflow-project
docker compose down --volumes --remove-orphans
  • Delete the project folder by running:
cd ~/Desktop
Remove-Item -Recurse -Force airflow-project
  • Confirm the folder is gone by running ls and checking that airflow-project no longer appears.

That's a Wrap!

That's a Wrap!

Nice work! You've gone from zero Airflow experience to running a scheduled, monitored ETL pipeline with full visibility into every task.

You've learned how to:

  • Set up a full Apache Airflow environment using Docker Compose with webserver, scheduler, and worker containers.
  • Write a Python DAG with an ETL pipeline that passes data between tasks using XCom.
  • Add cron scheduling and automatic retry logic so failed tasks recover without manual intervention.
  • Added parallel task branches for data validation that block the load step when quality checks fail.

Ready to quiz yourself?