Deep Dive into Apache Airflow

Build and break a 7-container Airflow stack to master DAGs and infrastructure.

Introduction

30 Second Summary

Every time you order food through an app, dozens of tasks fire in sequence behind the scenes. Something has to orchestrate those steps in the right order, retry when things fail, and route work differently depending on what's happening.

In this project, you will build an order processing pipeline on Apache Airflow, a workflow orchestration tool that coordinates complex multi-step processes. You'll run 7 interconnected containers, write tasks that branch and retry on their own, and deliberately break pieces to understand what each component does.

What You'll Build

You'll trigger your pipeline from a web dashboard, watch tasks execute across distributed workers in real time, and then pull individual services offline to see exactly what breaks.

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

  • A running 7-container Airflow stack where you can stop any service and immediately see what stops working in the system.
  • A TaskFlow API pipeline that you trigger from the web UI and watch branch, retry, and recover on its own.
  • Observable proof that large data in XCom bloats the metadata database, with a production-ready file-reference pattern already deployed.
  • Secret Mission: Process a variable number of order batches in parallel using dynamic task mapping.

Are there any prerequisites?

This project assumes familiarity with Python and basic Docker concepts (containers, images, volumes). Everything runs inside Docker containers, so you won't need to install Airflow on your machine directly.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters.

Set Up the Airflow Docker Compose Environment

To understand how Apache Airflow works in production, you need to see its components running as separate, independent services. A single-process install hides the distributed architecture that matters when things go wrong.

In this step, you will spin up a full CeleryExecutor stack using Docker Compose. This gives you 7 containers, each responsible for a distinct piece of the pipeline orchestration system.

In this step, get ready to:
  • Download and configure the official Airflow Docker Compose setup.
  • Initialize the Airflow metadata database.
  • Start all 7 services and verify the stack is healthy.
Download and initialize the Docker Compose environment

Airflow's official Docker Compose file defines all the services you need. You will download it, create the supporting directories, and initialize the database that Airflow uses to track every DAG run and task state.

Check your Docker Desktop memory

Airflow's containers need at least 4 GB of memory allocated to Docker Desktop. Open Docker Desktop, go to Settings then Resources, and confirm the memory slider is set to 4 GB or higher.

If it is lower, increase it and click Apply & Restart.

  • Open your terminal by pressing Cmd+Space, typing Terminal, and pressing Enter.
  • Create your project directory and move into it by running this command:
mkdir ~/airflow-deep-dive && cd ~/airflow-deep-dive

What does this command do?

This creates a folder called airflow-deep-dive in your home directory and immediately moves your terminal session into it. All subsequent commands will run from inside this folder.

  • Download the official Airflow 3.3.0 Docker Compose file by running:
curl -LfO 'https://airflow.apache.org/docs/apache-airflow/3.3.0/docker-compose.yaml'
  • Confirm the file downloaded by running:
ls

You should see docker-compose.yaml listed in the output.

Don't see docker-compose.yaml?

Make sure you are inside the ~/airflow-deep-dive directory. Run pwd to check your current location. If you are somewhere else, run cd ~/airflow-deep-dive and try the curl command again.

Help me troubleshoot the download

  • Create the required subdirectories and environment file by running:
mkdir -p ./dags ./logs ./plugins ./config && echo -e "AIRFLOW_UID=50000" > .env
  • Verify the directories and .env file were created by running:
ls -a

You should see dags, logs, plugins, config, .env, and docker-compose.yaml in the listing.

What are these directories for?

  • dags is where your pipeline code lives. Airflow's dag-processor watches this folder for new or changed files.
  • logs stores task execution logs so you can inspect what happened during each run.
  • plugins holds custom extensions to Airflow (you will not need it in this project, but Airflow expects it to exist).
  • config is for custom Airflow configuration overrides.
  • .env sets AIRFLOW_UID=50000 so the containers run with the correct user ID for file permissions.
  • Initialize the Airflow metadata database by running:
docker compose up airflow-init

This command pulls the required Docker images and sets up the Postgres database that Airflow uses to store DAG definitions, task states, and XCom values. It will take a minute or two depending on your internet speed.

Wait for the output to show exit code 0. You will see a line containing exited with code 0 at the end.

Seeing a non-zero exit code?

If you see exit code 1 or an error about insufficient memory, confirm Docker Desktop has at least 4 GB allocated. Increase the memory, restart Docker Desktop, then run docker compose up airflow-init again.

If you see an error about ports already in use, another service may be running on port 5432 (Postgres) or 6379 (Redis). Stop the conflicting service and retry.

Help me fix the init error

Start all services and verify health

The database is ready. Now you will start the full Airflow stack and confirm all 7 services are running and healthy.

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

This launches all containers in detached mode. The services need about 60 seconds to fully initialize and report healthy status.

  • Wait about 60 seconds, then check the container status by running:
docker compose ps

Before you look at the output, think about this: how many containers do you expect to see running, and what role does each one play?

You should see 7 containers, all showing a healthy status:

  • postgres stores all metadata (DAG definitions, task history, XCom values).
  • redis acts as the message broker that queues tasks for workers.
  • airflow-scheduler decides when tasks should run and queues them.
  • airflow-dag-processor parses your DAG files and updates the metadata database.
  • airflow-api-server serves the web UI and REST API.
  • airflow-worker picks up queued tasks from Redis and executes them.
  • airflow-triggerer handles deferrable tasks that wait for external events.

Why is this called a CeleryExecutor stack?

The CeleryExecutor is Airflow's most common production executor. It uses Redis as a task queue and separate worker processes to execute tasks. This separation means you can scale workers independently of the scheduler.

The alternative (airflow standalone) runs everything in one process. That hides exactly the distributed behavior you want to understand.

Some containers showing unhealthy?

If containers show as unhealthy or keep restarting, they may need more time. Wait another 30 seconds and run docker compose ps again.

If the worker or scheduler keeps crashing, check logs with docker compose logs airflow-scheduler to see the specific error.

Help me fix unhealthy containers

  • Open your browser and navigate to http://localhost:8080.
  • Log in with username airflow and password airflow.

You should see the Airflow UI dashboard with a list of example DAGs. This confirms the api-server, database, and dag-processor are all working together.

Can't reach localhost:8080?

The api-server may still be starting up. Wait another 30 seconds and refresh the page.

If it still does not load, run docker compose ps and check that airflow-api-server shows as healthy. If it shows as unhealthy, run docker compose logs airflow-api-server to find the error.

Help me access the Airflow UI

Your full Airflow environment is up and running. Next up, you will write your first DAG and watch it execute across this distributed stack.

Write and Trigger Your First TaskFlow DAG

Your 7-container Airflow stack is running and healthy. Now you need proof that the entire pipeline works end-to-end, from file parsing all the way to task execution.

The fastest way to prove it is to write a simple ETL (Extract, Transform, Load) DAG and trigger it from the UI. You will see the Graph view light up, inspect XCom data stored in the metadata database, and read task logs.

In this step, get ready to:
  • Write a three-task TaskFlow API DAG using @dag and @task decorators.
  • Trigger the DAG from the Airflow UI and observe all tasks complete successfully.
  • Inspect XCom values and task logs to confirm data flows between tasks.
Create the initial DAG file

Airflow's dag-processor watches the dags/ directory for Python files containing DAG definitions. You will create a single file that defines three tasks wired together as a pipeline.

  • Create the file dags/order_pipeline.py inside your airflow-deep-dive directory by running:
touch ~/airflow-deep-dive/dags/order_pipeline.py
  • Confirm the file was created by running:
ls ~/airflow-deep-dive/dags/

You should see order_pipeline.py listed in the output.

  • Open dags/order_pipeline.py in your code editor and add the following code to set up the DAG shell and the extract task:
import json
import os
import random

import pendulum

from airflow.sdk import dag, task


@dag(
    schedule=None,
    start_date=pendulum.datetime(2024, 1, 1, tz="UTC"),
    catchup=False,
    tags=["tutorial", "deep_dive"],
)
def order_processing_pipeline():
    """Order processing pipeline demonstrating TaskFlow, branching, and retries."""

    @task()
    def extract():
        """Extract order data (naive XCom approach for initial version)."""
        orders = {
            "1001": 301.27,
            "1002": 433.21,
            "1003": 502.22,
            "1004": 125.89,
            "1005": 678.50,
        }
        return orders

What does this code do?

  • from airflow.sdk import dag, task imports the TaskFlow API decorators. This is the canonical import path for Airflow 3.3.0.
  • @dag(schedule=None, ...) defines when the DAG runs. schedule=None means it only runs when you manually trigger it.
  • The extract task returns a dictionary of order data directly. Airflow automatically stores this return value in the metadata database via XCom so downstream tasks can access it.
  • Below the extract function (still inside order_processing_pipeline), add the remaining tasks and the pipeline wiring:
    @task(multiple_outputs=True)
    def transform(orders: dict):
        """Compute order summary statistics."""
        total_value = sum(orders.values())
        order_count = len(orders)
        return {
            "total_value": total_value,
            "order_count": order_count,
        }

    @task()
    def load(total_value: float, order_count: int):
        """Print results."""
        print(f"Total value: {total_value}, Order count: {order_count}")

    orders = extract()
    summary = transform(orders)
    load(summary["total_value"], summary["order_count"])


order_processing_pipeline()

What does this code do?

  • multiple_outputs=True on transform tells Airflow to split the returned dictionary into separate XCom entries. One for total_value, one for order_count.
  • The load task receives individual values rather than the whole dictionary. This is cleaner than unpacking inside the function.
  • The last three lines wire the tasks: extract() feeds transform(), which feeds load(). Airflow infers the dependency graph from these function calls.
  • order_processing_pipeline() at the bottom registers the DAG with Airflow. Without this call, the dag-processor would not detect it.
  • Save dags/order_pipeline.py.
  • Wait about 30 seconds for the dag-processor to detect the new file.
  • Switch back to the Airflow UI at http://localhost:8080. You should see order_processing_pipeline appear in the DAG list.

Don't see the DAG in the list?

  • Make sure your file is saved inside ~/airflow-deep-dive/dags/ (not a nested subfolder). The Docker volume mounts this directory to /opt/airflow/dags inside the containers.
  • Check for Python syntax errors by running python3 ~/airflow-deep-dive/dags/order_pipeline.py in your local terminal. If there is a typo, the dag-processor silently skips the file.
  • Confirm the dag-processor is still running with docker ps and check its status shows healthy.

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

✔️ Awesome, I've got everything!

Great. Double check you have saved your file before moving on.

ⓧ I'd like to double check the full code

Your dags/order_pipeline.py file should contain two chunks, in order:

  • Chunk 1: The imports, @dag decorator, order_processing_pipeline function definition, and the extract task that returns the orders dictionary.
  • Chunk 2: The transform task (with multiple_outputs=True), the load task, the three wiring lines (orders = extract() through load(...)), and the final order_processing_pipeline() call.

Scroll up and confirm each chunk matches what you pasted. The file should be about 50 lines total.

Trigger and observe in the UI

The DAG exists in Airflow's metadata, but it has not run yet. You need to unpause it and trigger a manual run to see the full execution flow.

  • In the Airflow UI, find order_processing_pipeline in the DAG list.
  • Click the pause/unpause toggle next to the DAG name to set it to the unpaused (active) state.
  • Click on the DAG name to open its detail page.
  • Click the play icon (Trigger DAG button) to start a manual run.
  • In the trigger dialog, click Trigger without changing any parameters.

Before you check the result, predict: which task do you think will finish first, and how will data flow between them?

  • Switch to the Graph view.
  • Wait a few seconds for all three task boxes (extract, transform, load) to turn green, indicating success.

All three tasks run in sequence because each one depends on the output of the previous one.

  • Click on the transform task box in the Graph view.
  • Navigate to the XCom tab in the task instance details.
  • You should see two entries: total_value with the value 2041.09 and order_count with the value 5.

Why are there separate XCom entries?

The multiple_outputs=True parameter told Airflow to split the dictionary into individual XCom keys. This is why downstream tasks can pull summary["total_value"] and summary["order_count"] as separate values rather than receiving the whole dictionary.

XCom stores these values in the Postgres metadata database. This is fine for small payloads like numbers and short strings.

  • Click on the load task box in the Graph view.
  • Open the Logs tab to see the task's printed output.
  • You should see the line Total value: 2041.09, Order count: 5 in the log output.

Tasks stuck in queued or failed?

  • If tasks stay in a "queued" state (orange), confirm the worker is running with docker ps. The worker picks tasks off the Redis queue and executes them.
  • If a task shows "failed" (red), click on it and check the Logs tab. The error message will point to the issue, usually a syntax error in the DAG file.
  • If the DAG ran but XCom values are missing, make sure multiple_outputs=True is set on the transform task decorator.

Need more help? Help me debug why my Airflow tasks are stuck or failing.

Your pipeline is running end-to-end. Every piece of the stack contributed: the dag-processor parsed your file, the scheduler queued the tasks, Redis delivered them, and the worker executed each one in order.

Next up, you will push this pipeline to its limits by passing a large payload through XCom and discover why that is an anti-pattern in production.

Hit the XCom Limitation and Refactor

Your pipeline runs end-to-end, but there is a hidden problem. The extract task returns its entire order dictionary through XCom, which stores that data directly in the Postgres metadata database.

With 5 orders, the stored blob is tiny. But what happens when you scale to 10,000 orders? Every DAG run, every retry, every historical record stores that entire payload in the database. In this step, you will scale the data to expose the problem firsthand, then refactor to the production pattern.

In this step, get ready to:
  • Scale extract to generate 10,000 orders and observe the XCom bloat.
  • Refactor to the file-reference pattern used in production.
Make extract return a large payload and observe the problem

The current extract task returns a hardcoded dictionary of 5 orders. You are going to scale that up to 10,000 randomly generated orders to simulate a real dataset, while keeping the same naive return pattern.

  • In dags/order_pipeline.py, find the extract function body.
  • Find this block of code inside the function:
        orders = {
            "1001": 301.27,
            "1002": 433.21,
            "1003": 502.22,
            "1004": 125.89,
            "1005": 678.50,
        }
        return orders
  • Replace it with this single line that generates 10,000 random orders and still returns them directly:
        orders = {str(i): round(random.uniform(10, 500), 2) for i in range(10000)}
        return orders

What does this change do?

The dict comprehension generates 10,000 key-value pairs with random order amounts between $10 and $500. The function still returns the entire dictionary directly, so XCom stores all 10,000 entries in the Postgres metadata database.

  • Save dags/order_pipeline.py.
  • Wait about 30 seconds for the dag-processor to detect the file change.
  • In the Airflow UI at http://localhost:8080, click on the order_processing_pipeline DAG.
  • Click Trigger DAG to start a new run.
  • Wait for all tasks to show green (success).

Before you inspect the XCom tab, predict: what do you think is stored in the metadata database for the extract task's return value?

  • Click on the extract task instance in the Graph view.
  • Navigate to the XCom tab.

You see a massive JSON blob containing all 10,000 order entries stored directly in the metadata database. That is the anti-pattern in action.

Why is this a problem?

XCom is "only designed for small amounts of data" according to the Airflow docs. Every retry and every historical run stores this entire blob in Postgres.

At scale, this fills your metadata database and slows the scheduler. The production pattern is to pass references (file paths, S3 keys, database table names) through XCom and store the actual data elsewhere.

DAG not updating after saving?

The dag-processor needs about 30 seconds to detect file changes. If the DAG still shows the old version after a minute, check that you saved the file in the correct directory (~/airflow-deep-dive/dags/).

If the DAG shows an import error in the UI, check for syntax errors in your dict comprehension.

Help me debug my DAG file changes not being detected by the dag-processor.

Refactor to the file-based reference pattern

Now you will fix the anti-pattern. Instead of returning the data directly, the extract task will write to a shared file and return only the file path string through XCom.

  • In dags/order_pipeline.py, replace the entire extract function (from @task() to return orders) with this refactored version:
    @task()
    def extract():
        """Extract order data and write to shared storage."""
        orders = {str(i): round(random.uniform(10, 500), 2) for i in range(10000)}
        data_path = "/opt/airflow/dags/data/orders.json"
        os.makedirs(os.path.dirname(data_path), exist_ok=True)
        with open(data_path, "w") as f:
            json.dump(orders, f)
        return data_path

What does this code do?

  • os.makedirs creates the data/ directory inside the DAGs folder if it does not exist yet.
  • json.dump writes the 10,000 orders to a JSON file on the shared filesystem.
  • The function returns only the file path string (/opt/airflow/dags/data/orders.json) through XCom. The actual data lives on disk, not in Postgres.
  • Now replace the transform function. Find it below the extract function you just replaced, then swap the entire function with this:
    @task(multiple_outputs=True)
    def transform(data_path: str):
        """Compute order summary statistics from file."""
        with open(data_path, "r") as f:
            orders = json.load(f)
        total_value = sum(orders.values())
        order_count = len(orders)
        return {
            "total_value": total_value,
            "order_count": order_count,
        }

What changed in transform?

The transform function now accepts data_path: str instead of orders: dict. It reads the orders from the file at that path instead of receiving the data through XCom directly.

  • Also update the wiring at the bottom of the DAG function. Find these lines:
    orders = extract()
    summary = transform(orders)
    load(summary["total_value"], summary["order_count"])
  • Replace them with:
    data_path = extract()
    summary = transform(data_path)
    load(summary["total_value"], summary["order_count"])

Why rename the variable?

The variable changes from orders to data_path because extract now returns a path string, not data. The load function stays unchanged since it still receives the computed summary values.

  • Save dags/order_pipeline.py.
  • Wait about 30 seconds for the dag-processor to pick up the change.

Before you trigger the DAG, predict: what do you expect to see in the XCom tab for the extract task this time?

  • Click Trigger DAG to start a new run.
  • Wait for all tasks to show green (success).
  • Click on the extract task instance in the Graph view.
  • Navigate to the XCom tab.

The XCom tab now shows only the string /opt/airflow/dags/data/orders.json instead of a massive JSON blob. The actual data lives on the shared filesystem, and only a lightweight reference passes through the metadata database.

Seeing a task failure after refactoring?

Check the task logs for the failing task. Common causes include a typo in the data_path variable name (it must match exactly between extract and transform), or forgetting to update the wiring from orders = extract() to data_path = extract().

Help me debug my refactored DAG that is failing after I changed the extract function to return a file path.

✔️ Awesome, I've got everything!

Great. Double check you have saved dags/order_pipeline.py and confirmed the XCom tab shows only the file path string.

ⓧ I'd like to double check the full code

Your dags/order_pipeline.py should contain these sections in order. Scroll up and confirm each one is present:

  • Imports: json, os, random, pendulum, and from airflow.sdk import dag, task.
  • The @dag decorator with schedule=None, catchup=False, and the order_processing_pipeline function definition.
  • The extract function that generates 10,000 orders, writes them to /opt/airflow/dags/data/orders.json, and returns only the path string.
  • The transform function that accepts data_path: str and reads from the file.
  • The load function (unchanged from before).
  • Wiring: data_path = extract(), summary = transform(data_path), load(summary["total_value"], summary["order_count"]).
  • The order_processing_pipeline() call at the very bottom of the file.

Your pipeline now follows the production pattern for data passing. Next up, you will add branching logic that routes processing based on the order total, plus retry handling that automatically recovers from transient failures.

Add Branching and Retry Logic

Your pipeline processes every order batch the same way right now. Real pipelines need conditional routing. High-value batches might require finance team alerts, while normal batches only need a standard report.

Airflow's @task.branch() decorator lets a task decide at runtime which downstream path to follow. Combined with automatic retries, you get a pipeline that routes intelligently and recovers from transient failures without you writing retry loops.

In this step, get ready to:
  • Add a branching task that routes processing based on order value.
  • Add retry logic that automatically recovers from simulated failures.
  • Observe branching and retries in the Airflow UI.
Add a branching decision task

Right now your pipeline runs extract, transform, then load in a straight line. You need a decision point that checks the total order value and routes to different tasks depending on the result.

First, replace the random 10,000-order extract with a fixed set of 5 orders. Fixed data gives you predictable totals, which makes it easy to verify your branch is routing correctly.

  • In dags/order_pipeline.py, find the extract function.
  • Replace the entire function body (everything inside def extract():) with this fixed order data:
    @task()
    def extract():
        """Extract order data and write to shared storage."""
        orders = {
            "1001": 301.27,
            "1002": 433.21,
            "1003": 502.22,
            "1004": 125.89,
            "1005": 678.50,
        }
        data_path = "/opt/airflow/dags/data/orders.json"
        os.makedirs(os.path.dirname(data_path), exist_ok=True)
        with open(data_path, "w") as f:
            json.dump(orders, f)
        return data_path

Why fixed orders?

These 5 orders sum to 2041.09, which is above the 1500 threshold you are about to set. That means the branch always takes the high-value path, making it easy to verify.

In a real pipeline you would keep dynamic data. Fixed data here just makes the branching behavior predictable while you learn the pattern.

Now remove the old load function and its wiring, then add three new functions: a branch decision, a high-value alert, and a standard report generator.

  • Delete the entire load function (the @task() decorator, def load(...):, and its body).
  • Delete the three wiring lines at the bottom of the DAG function (data_path = extract(), summary = transform(data_path), and load(summary["total_value"], summary["order_count"])).
  • After the transform function, add the following new functions and pipeline wiring:
    @task.branch()
    def check_threshold(total_value: float):
        """Route to different processing based on order total."""
        if total_value > 1500:
            return "alert_finance"
        return "generate_report"

    @task()
    def alert_finance():
        """Alert finance team for high-value batches."""
        print("Finance team notified of high-value batch!")

    @task()
    def generate_report():
        """Generate standard batch report for normal-value batches."""
        print("Standard batch report generated and filed.")

    # Wire the pipeline
    data_path = extract()
    summary = transform(data_path)
    branch = check_threshold(summary["total_value"])
    branch >> [alert_finance(), generate_report()]

What does this code do?

check_threshold uses the @task.branch() decorator. Its return value is a task ID string. Airflow executes only the task whose ID matches the returned string and skips the other.

The >> operator wires the branch to both possible downstream tasks. At runtime, only one path will actually execute.

  • Save dags/order_pipeline.py.
  • Wait about 30 seconds for the dag-processor to pick up your changes.

Before you trigger the DAG, think about this: your 5 orders sum to 2041.09. The threshold is 1500. Which branch do you expect to run, and which do you expect to be skipped?

  • In the Airflow UI, find the order_processing_pipeline DAG and click Trigger DAG.
  • Switch to the Graph view.

You should see alert_finance in green (success) and generate_report in pink (skipped). Since 2041.09 exceeds 1500, Airflow routed to alert_finance and skipped the other path.

Both tasks showing as skipped?

Check that check_threshold returns the exact string "alert_finance" or "generate_report". The returned string must match the Python function name exactly.

Also confirm you deleted the old load function and its wiring completely. Leftover references cause import errors that prevent the DAG from parsing.

help me debug why my branching task is skipping both paths

Add retry logic to simulate transient failures

In production, tasks fail for transient reasons all the time. A network timeout, a rate limit, a momentary service outage. Airflow can retry tasks automatically so you do not have to write retry loops yourself.

  • In dags/order_pipeline.py, find the alert_finance function.
  • Find this code:
    @task()
    def alert_finance():
        """Alert finance team for high-value batches."""
        print("Finance team notified of high-value batch!")
  • Replace it with this version that adds retries and a simulated random failure:
    @task(retries=3)
    def alert_finance():
        """Alert finance team for high-value batches. Simulates transient failures."""
        if random.random() < 0.4:
            raise Exception("Simulated connection timeout - will retry")
        print("Finance team notified of high-value batch!")

What does this code do?

retries=3 tells Airflow to retry this task up to 3 times if it raises an exception. The random.random() < 0.4 line gives a 40% chance of failure on each attempt, simulating a flaky external service.

Airflow handles the retry loop, wait time, and attempt counting for you. Each retry attempt gets its own log entry so you can trace what happened.

  • Save dags/order_pipeline.py.
  • Wait about 30 seconds for the dag-processor to detect the change.

Now trigger the DAG a few times to see the retry behavior in action. With a 40% failure rate, some runs will retry before succeeding.

  • Click Trigger DAG to start a new run.
  • In the Graph view, click on the alert_finance task instance.
  • Check the task instance details for the retry count.
  • Open the Logs tab to see each retry attempt logged separately.

If the task fails on its first attempt, you will see the retry counter increment. Airflow automatically re-queues the task until it either succeeds or exhausts all 3 retries.

Task fails after all retries?

That is expected sometimes. With a 40% failure rate and only 3 retries, there is roughly a 6% chance all attempts fail (0.4 to the power of 4). Trigger the DAG again and it will likely succeed on the next run.

help me understand why my alert_finance task keeps failing after all retries

✔️ Awesome, I've got everything!

Great. Double check you have saved dags/order_pipeline.py and that the Graph view shows your branching logic working correctly.

ⓧ I'd like to double check the full code

Here is the complete dags/order_pipeline.py file. Compare your code against this:

import json
import os
import random

import pendulum

from airflow.sdk import dag, task

Your file should have these sections in order after the imports:

  • The @dag decorator and def order_processing_pipeline(): function definition.
  • The extract function with 5 fixed orders writing to /opt/airflow/dags/data/orders.json.
  • The transform function with multiple_outputs=True.
  • The check_threshold function with @task.branch().
  • The alert_finance function with @task(retries=3) and the random failure simulation.
  • The generate_report function.
  • The pipeline wiring ending with branch >> [alert_finance(), generate_report()].
  • The order_processing_pipeline() call at the bottom of the file.
@dag(
    schedule=None,
    start_date=pendulum.datetime(2024, 1, 1, tz="UTC"),
    catchup=False,
    tags=["tutorial", "deep_dive"],
)
def order_processing_pipeline():
    """Order processing pipeline demonstrating TaskFlow, branching, and retries."""

    @task()
    def extract():
        """Extract order data and write to shared storage."""
        orders = {
            "1001": 301.27,
            "1002": 433.21,
            "1003": 502.22,
            "1004": 125.89,
            "1005": 678.50,
        }
        data_path = "/opt/airflow/dags/data/orders.json"
        os.makedirs(os.path.dirname(data_path), exist_ok=True)
        with open(data_path, "w") as f:
            json.dump(orders, f)
        return data_path

    @task(multiple_outputs=True)
    def transform(data_path: str):
        """Compute order summary statistics from file."""
        with open(data_path, "r") as f:
            orders = json.load(f)
        total_value = sum(orders.values())
        order_count = len(orders)
        return {
            "total_value": total_value,
            "order_count": order_count,
        }
    @task.branch()
    def check_threshold(total_value: float):
        """Route to different processing based on order total."""
        if total_value > 1500:
            return "alert_finance"
        return "generate_report"

    @task(retries=3)
    def alert_finance():
        """Alert finance team for high-value batches. Simulates transient failures."""
        if random.random() < 0.4:
            raise Exception("Simulated connection timeout - will retry")
        print("Finance team notified of high-value batch!")
    @task()
    def generate_report():
        """Generate standard batch report for normal-value batches."""
        print("Standard batch report generated and filed.")

    # Wire the pipeline
    data_path = extract()
    summary = transform(data_path)
    branch = check_threshold(summary["total_value"])
    branch >> [alert_finance(), generate_report()]


order_processing_pipeline()

Your pipeline now makes intelligent routing decisions and recovers from failures automatically. Next, you will break the infrastructure itself to understand what each component does when it is gone.

Break the Infrastructure to Understand It

You have a working DAG, and you know what each task does. But do you know what each infrastructure component does?

Reading that the scheduler triggers workflows or that the worker executes tasks is one thing. Watching the system degrade when you kill a component builds operational intuition you cannot get from documentation alone.

In this step, get ready to:
  • Stop the worker and observe tasks queuing but never executing.
  • Stop the scheduler and observe new runs failing to trigger.
  • Stop the dag-processor and observe the UI showing stale code.
Kill the worker and observe task queuing

The worker is the container that actually executes your task code. Without it, the scheduler can still queue tasks, and Redis can still hold messages, but nothing picks them up.

  • Return to the Airflow UI at http://localhost:8080.
  • Click on order_processing_pipeline and click Trigger DAG to confirm tasks execute normally (all green).

Now you have a baseline of the happy path. Time to break it.

  • Stop the worker by running this command in your terminal:
docker compose stop airflow-worker
  • Switch back to the Airflow UI.
  • Click Trigger DAG again to start a new run.
  • Switch to the Graph view and watch the tasks.

The tasks move to a queued state (orange) but never transition to running. The scheduler did its job by queuing them, and Redis is holding the messages, but nobody is on the other end to pick them up.

What is happening under the hood?

The CeleryExecutor architecture separates task scheduling from task execution. The scheduler pushes task messages onto the Redis queue. The worker pulls messages from Redis and runs the Python code.

With no worker running, the queue grows but nothing drains it. This is exactly what happens in production when a worker node crashes.

Before you restart the worker, what do you think will happen to those queued tasks?

  • Restart the worker by running:
docker compose start airflow-worker
  • Switch back to the Graph view in the Airflow UI.

The queued tasks immediately start executing. They transition from orange (queued) to green (success) as the worker drains the Redis queue.

Tasks still showing queued after restart?

Give it 10-15 seconds for the worker to fully reconnect to Redis. If tasks stay orange for more than 30 seconds, check the worker logs:

docker compose logs airflow-worker --tail 20

Look for connection errors to Redis. Help me debug why my Airflow worker isn't picking up queued tasks.

Kill the scheduler and observe no new runs

The scheduler is responsible for turning DAG runs into individual task instances and deciding when to queue them. Without it, existing queued tasks can still execute, but no new work gets scheduled.

  • Stop the scheduler by running:
docker compose stop airflow-scheduler
  • Switch to the Airflow UI.
  • Click Trigger DAG to attempt a new run.
  • Switch to the Graph view for this new run.

The trigger may appear to register, but tasks never move to the queued state. The scheduler is not processing the DAG run, so no task instances get created or queued.

Scheduler vs worker: what's the difference?

The scheduler decides WHAT should run and WHEN to queue it. The worker decides HOW to run it. Without the scheduler, the system has muscles (worker) but no brain telling them what to do.

What do you expect will happen when you bring the scheduler back online?

  • Restart the scheduler by running:
docker compose start airflow-scheduler
  • Watch the Airflow UI. Within a few seconds, the pending DAG run's tasks begin moving to queued, then running, then success.

The scheduler picks up where it left off, processing any DAG runs that were triggered while it was down.

DAG run doesn't resume after restart?

Wait up to 30 seconds for the scheduler to fully initialize and begin its scheduling loop. If the run still shows no progress, confirm the worker is healthy:

docker ps --filter "name=airflow"

All containers should show healthy status. Help me figure out why my scheduler isn't processing DAG runs.

Kill the dag-processor and observe stale DAGs

The dag-processor continuously parses your DAG files from disk and updates the metadata database. Without it, changes you make to Python files are invisible to the rest of Airflow.

  • Open dags/order_pipeline.py in your code editor.
  • Find the extract function's first line inside the function body.
  • Add this print statement as the very first line inside the function:
print("Extract task v2 - dag-processor test")
  • Save dags/order_pipeline.py.
  • Now immediately stop the dag-processor before it can detect the change:
docker compose stop airflow-dag-processor
  • In the Airflow UI, trigger the DAG and check the task logs for the extract task.

The logs show the OLD version of your code without the new print statement. The file on disk has changed, but the dag-processor never parsed the update into the metadata database.

Why can't the scheduler read the files directly?

In a distributed system, the scheduler and workers might run on different machines that don't share the same filesystem. The dag-processor's job is to parse DAG files and store the structure in the shared Postgres metadata database. Every other component reads from that database, not from disk.

  • Restart the dag-processor by running:
docker compose start airflow-dag-processor
  • Wait about 30 seconds for the dag-processor to parse the updated file.
  • Trigger the DAG again in the UI and check the extract task logs.

You now see the Extract task v2 - dag-processor test print output in the logs. The dag-processor detected the file change and updated the metadata database.

  • Now remove the test print statement from dags/order_pipeline.py and save the file to restore it to its original state.

Still seeing old code after restart?

The dag-processor needs up to 30 seconds to complete its parsing cycle after starting. Make sure you saved the file before checking.

Confirm the dag-processor is actually running:

docker ps --filter "name=dag-processor"

You should see it listed with a healthy status. Help me understand why my DAG file changes aren't showing up in the Airflow UI.

Verify all services are healthy

You have now broken and repaired each core component. Before you wrap up, confirm everything is back to a clean state.

What do you expect to see when you check the container health?

  • Verify all 7 containers are healthy by running:
docker ps --format "table {{.Names}}\t{{.Status}}"

You should see all 7 services listed with (healthy) in their status. The postgres, redis, scheduler, dag-processor, api-server, worker, and triggerer containers are all running.

What did you just prove?

  • The worker executes tasks. Without it, tasks queue forever.
  • The scheduler creates and queues task instances. Without it, no new work gets processed.
  • The dag-processor bridges your DAG files on disk to the metadata database. Without it, code changes are invisible to the system.

You now have firsthand operational intuition about Airflow's distributed architecture. Next up, a Secret Mission that pushes your pipeline further with dynamic task mapping.

Secret mission

Add Dynamic Task Mapping

Your current DAG processes a fixed set of orders. Real pipelines handle a variable number of batches that aren't known until runtime. Can you use Airflow's dynamic task mapping to fan out processing across multiple order batches in parallel?

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 ongoing cloud costs. However, the 7 Airflow containers consume approximately 4GB of RAM while running.

Resources you used:

  • Docker containers (7 services: postgres, redis, airflow-scheduler, airflow-dag-processor, airflow-api-server, airflow-worker, airflow-triggerer).
  • Docker volumes for Postgres metadata and Redis data.
  • Local project directory: ~/airflow-deep-dive (including dags, logs, plugins, config subdirectories).

Keep everything running

No action needed. Choose this if you plan to keep experimenting with your DAGs or adding new pipelines to the environment.

  • Your containers will keep running in the background and persist across Docker Desktop restarts.
  • The Airflow UI stays accessible at http://localhost:8080.
  • Any new DAG files you add to ~/airflow-deep-dive/dags will be picked up automatically by the dag-processor.

Pause - I'll come back to this later

Shut down running containers to free up memory while keeping all your data and DAG files intact. You can resume everything later without re-initializing.

  • Stop all containers by running this command from your ~/airflow-deep-dive directory:
cd ~/airflow-deep-dive
docker compose stop
  • When you want to resume, start everything back up with:
cd ~/airflow-deep-dive
docker compose start
  • Your DAG files, metadata database, and all run history will be exactly as you left them.

Delete - I don't want to use this again

Remove all containers, volumes, downloaded images, and the project directory entirely. This frees up disk space and RAM.

  • Remove all Docker containers, volumes, and images by running:
cd ~/airflow-deep-dive
docker compose down --volumes --rmi all
  • Delete the project directory and all its contents:
rm -rf ~/airflow-deep-dive
  • Confirm the directory is gone by running:
ls ~/airflow-deep-dive
  • You should see a "No such file or directory" message, confirming everything has been removed.

That's a Wrap!

That's a Wrap!

Nice work! You've built and operated a full production-patterned Apache Airflow environment from the ground up.

You've learned how to:

  • Deploy a CeleryExecutor Airflow stack with 7 interconnected services in Docker Compose, and identify what each component does by breaking it.
  • Write a TaskFlow API DAG that passes data through file references instead of bloating the metadata database, routes processing through conditional branches with @task.branch(), and recovers from transient failures automatically with @task(retries=3).
  • Build operational intuition by stopping the worker, scheduler, and dag-processor individually and observing how the system degrades and recovers.
  • Secret Mission: Implemented dynamic task mapping with .expand() to process a variable number of order batches in parallel at runtime.

Ready to quiz yourself?