Build a Pipeline Schedule Optimizer

Use OR-Tools CP-SAT to auto-schedule Spark pipelines with constraints.

Introduction

30 Second Summary

If you manage data pipelines, you know the pain of Monday morning resource spikes. Every team schedules their Spark jobs to run "first thing," and suddenly your cluster is overloaded while Tuesday sits nearly idle. Manual cron tuning breaks down the moment someone adds a new pipeline or changes a dependency.

In this project, you will build a command-line optimizer that uses constraint programming to schedule your pipelines across time slots. It respects dependency order, stays within cluster capacity, and produces a visual before/after comparison you can show your boss.

What You'll Build

You run a single command, and a chart pops up showing your pipelines spread evenly across the week with zero dependency violations and no resource spikes. You send that chart to your boss as proof that a solver can replace hours of manual schedule tuning.

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

  • A working optimizer where you feed in a JSON config describing 8 pipelines with dependencies and resource costs, and the OR-Tools CP-SAT solver returns a valid schedule in seconds.
  • A Gantt-style chart with resource utilization bars showing the difference between "everything piled on Monday" and the optimized spread.
  • A solver that guarantees no pipeline starts before its upstream dependency finishes, and peak concurrent resource usage never exceeds your defined cluster capacity.
  • Secret Mission: Generate Airflow-compatible cron expressions and a DAG file from the optimized schedule.

Are there any prerequisites?

This project assumes you're comfortable writing Python and familiar with pipeline or DAG concepts. Experience with Airflow scheduling is helpful for context but not required to complete the steps.

Before We Start

Before We Start...

Before we dive in, let's lock in what you're about to build and why it matters. When a team has three Spark pipelines, a human can juggle their cron schedules in a spreadsheet. When that number climbs to ten or twenty with tangled dependencies and shared cluster resources, manual scheduling breaks down into dependency violations and Monday-morning resource spikes.

Set Up OR-Tools and Project Structure

Before you can model pipeline constraints or find optimal schedules, you need two things: a solver engine that can crunch the math, and a charting library that can show the results.

In this step, you'll install Google's OR-Tools constraint solver and matplotlib for visualization, then create the project files you'll build on in later steps.

In this step, get ready to:
  • Create a project folder and Python virtual environment.
  • Install OR-Tools and matplotlib via pip.
  • Create the project files and verify everything imports correctly.
Create the project directory and virtual environment

You need an isolated Python environment so your project's dependencies don't conflict with anything else on your machine.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:

macOS

mkdir pipeline-optimizer && cd pipeline-optimizer

Windows

mkdir pipeline-optimizer; cd pipeline-optimizer
  • Confirm you're in the right place by running:
pwd

You should see a path ending in pipeline-optimizer.

  • Create a virtual environment and activate it by running these commands:

macOS

python3 -m venv venv
source venv/bin/activate

Windows

python -m venv venv
venv\Scripts\activate

Your terminal prompt now starts with (venv). This confirms the virtual environment is active.

Don't see (venv) in your prompt?

Make sure you ran the activation command (not just the python3 -m venv venv line). The venv creation and activation are two separate steps.

On macOS, confirm the venv/bin/activate file exists by running ls venv/bin/. On Windows, check venv\Scripts\ for the activate script.

help me activate my Python virtual environment

Install dependencies
  • Install OR-Tools and matplotlib by running:
pip install ortools==9.15.6755 matplotlib

You'll see pip downloading and installing packages. The final lines should show Successfully installed followed by a list of package names.

Why pin the OR-Tools version?

Pinning to 9.15.6755 ensures the API calls in this project work exactly as written. OR-Tools occasionally changes method names between releases, so a fixed version avoids surprises.

Seeing a network error or timeout?

Check your internet connection and try the install command again. If pip reports a version conflict, make sure your (venv) prefix is visible. Installing inside the virtual environment avoids conflicts with system packages.

help me fix my pip install error for ortools

Create project files and verify installation

Your project needs three files: one for pipeline configuration, one for the solver logic, and one for chart output.

  • Create the three project files by running:

macOS

touch pipelines.json optimizer.py visualize.py

Windows

New-Item -ItemType "File" -Path "pipelines.json", "optimizer.py", "visualize.py"
  • Confirm the files exist by running:
ls

You should see pipelines.json, optimizer.py, visualize.py, and the venv folder listed in the output.

  • Open VS Code with your project folder by pressing Cmd+Space (macOS) or the Windows key (Windows) to open search.
  • Type VS Code and press Enter to open it.
  • Click File in the top menu bar.
  • Click Open Folder.
  • Select the pipeline-optimizer folder on your Desktop, then click Open.

You should see pipelines.json, optimizer.py, and visualize.py in the VS Code Explorer sidebar on the left.

Now for the final check. Before you run the next command, what do you think will happen if OR-Tools installed correctly?

  • Back in your terminal, verify OR-Tools imports correctly by running:
python -c "from ortools.sat.python import cp_model; print('OR-Tools ready')"

You should see OR-Tools ready printed. This confirms the CP-SAT solver module is installed and importable.

Seeing ModuleNotFoundError?

Make sure your terminal still shows the (venv) prefix. If it disappeared, reactivate the virtual environment and try the verification command again.

If the prefix is there but the import still fails, run pip list and check that ortools appears in the output. If not, re-run the install command from the previous substep.

help me fix the OR-Tools import error

Your environment is ready. Next up, you'll define your pipeline configuration and give the optimizer something to work with.

Define Your Pipeline Configuration

Your environment is ready with OR-Tools and matplotlib installed. Now the optimizer needs something to optimize.

Every constraint solver requires structured input that describes the problem. In this step, you will define your Spark pipelines as a JSON configuration and write the Python code that loads, validates, and naively schedules them.

In this step, get ready to:
  • Define 8 pipelines with durations, resource demands, and dependency chains in a JSON file.
  • Write a loader that validates all dependency references before the solver touches them.
  • Generate a naive baseline schedule that exposes the resource-spike problem.
Create the pipeline JSON configuration

The optimizer needs to know each pipeline's name, how long it runs, how many cluster slots it consumes, and which other pipelines must finish first. A JSON file keeps this configuration separate from the solver logic so you can swap pipeline sets without editing code.

  • Open pipelines.json in VS Code and paste the following configuration:
{
  "cluster_capacity": 5,
  "horizon": 40,
  "pipelines": [
    {"name": "raw_ingestion", "duration": 3, "resource_demand": 2, "dependencies": []},
    {"name": "user_events", "duration": 2, "resource_demand": 1, "dependencies": []},
    {"name": "transaction_load", "duration": 4, "resource_demand": 2, "dependencies": []},
    {"name": "data_transform", "duration": 3, "resource_demand": 2, "dependencies": ["raw_ingestion"]},
    {"name": "user_profiles", "duration": 2, "resource_demand": 1, "dependencies": ["user_events"]},
    {"name": "transaction_agg", "duration": 3, "resource_demand": 2, "dependencies": ["transaction_load"]},
    {"name": "ml_feature_store", "duration": 4, "resource_demand": 3, "dependencies": ["data_transform", "user_profiles"]},
    {"name": "dashboard_refresh", "duration": 2, "resource_demand": 1, "dependencies": ["transaction_agg", "ml_feature_store"]}
  ]
}

What does this configuration describe?

  • cluster_capacity: 5 means at most 5 resource slots can run concurrently across the whole cluster.
  • horizon: 40 sets the scheduling window to 40 time slots (the solver won't schedule anything beyond this).
  • Each pipeline has a duration (how many slots it occupies), a resource_demand (how many cluster slots it consumes while running), and a dependencies list (which pipelines must finish before it can start).
  • The dependency chains form a DAG. For example, ml_feature_store cannot start until both data_transform and user_profiles finish.
  • Save pipelines.json.
Load and validate the configuration in optimizer.py

Before the solver can reason about these pipelines, your code needs to read the JSON and confirm every dependency reference actually points to an existing pipeline name. A typo in a dependency would silently produce a wrong schedule.

  • Open optimizer.py and paste the imports and load_pipelines() function:
import json
import collections
from ortools.sat.python import cp_model


def load_pipelines(config_path="pipelines.json"):
    # Read the pipeline configuration file
    with open(config_path) as f:
        config = json.load(f)

    pipelines = config["pipelines"]
    capacity = config["cluster_capacity"]
    horizon = config["horizon"]

    # Validate that every dependency references an existing pipeline
    names = {p["name"] for p in pipelines}
    for p in pipelines:
        for dep in p["dependencies"]:
            if dep not in names:
                raise ValueError(
                    f"Pipeline '{p['name']}' depends on unknown pipeline '{dep}'"
                )

    return pipelines, capacity, horizon

What does this code do?

  • The imports bring in json for config parsing, collections for a data structure you will use in the next step, and cp_model from OR-Tools for the constraint solver.
  • load_pipelines() reads the JSON, extracts the pipeline list, cluster capacity, and time horizon.
  • The validation loop builds a set of all pipeline names, then checks every dependency reference against that set. A typo like "raw_ingest" instead of "raw_ingestion" raises an error immediately.
  • Below load_pipelines(), add the main() function that loads and prints the pipeline summary:
def main():
    pipelines, capacity, horizon = load_pipelines()

    print(f"Loaded {len(pipelines)} pipelines (capacity: {capacity}, horizon: {horizon})")
    print("\nPipelines:")
    for p in pipelines:
        deps = ", ".join(p["dependencies"]) if p["dependencies"] else "none"
        print(f"  {p['name']:<20} duration={p['duration']} demand={p['resource_demand']} deps=[{deps}]")


if __name__ == "__main__":
    main()

What does this code do?

main() calls load_pipelines() and prints a formatted summary of each pipeline's name, duration, resource demand, and dependencies. The if __name__ guard ensures this only runs when you execute the file directly.

  • Save optimizer.py.
  • Run the script to confirm the configuration loads correctly:
python optimizer.py

You should see all 8 pipelines listed with their durations, demands, and dependency references.

Seeing a FileNotFoundError?

  • Make sure pipelines.json is saved in the same folder as optimizer.py. Both files must be in pipeline-optimizer/.
  • Check that your terminal is inside the pipeline-optimizer directory and the virtual environment is active (you should see (venv) in your prompt).
  • help me debug a FileNotFoundError when running optimizer.py
Add a naive baseline schedule

To prove the optimizer adds value, you need a "before" picture. The naive schedule starts every pipeline as early as possible respecting only dependency order, completely ignoring resource limits. This mimics what happens when engineers hand-schedule everything for Monday morning.

  • In optimizer.py, add the build_naive_schedule() function below load_pipelines():
def build_naive_schedule(pipelines):
    # Schedule everything ASAP respecting only dependency order
    schedule = {}
    resolved = set()

    def get_earliest_start(pipeline):
        earliest = 0
        for dep in pipeline["dependencies"]:
            dep_end = schedule[dep]["start"] + schedule[dep]["duration"]
            earliest = max(earliest, dep_end)
        return earliest

    # Topological sort via repeated passes
    remaining = list(pipelines)
    while remaining:
        progress = False
        for p in list(remaining):
            if all(dep in resolved for dep in p["dependencies"]):
                start = get_earliest_start(p)
                schedule[p["name"]] = {
                    "start": start,
                    "duration": p["duration"],
                    "resource_demand": p["resource_demand"],
                }
                resolved.add(p["name"])
                remaining.remove(p)
                progress = True
        if not progress:
            raise ValueError("Circular dependency detected")

    return schedule

What does this code do?

  • get_earliest_start() looks at a pipeline's dependencies and finds the latest end time among them. The pipeline cannot start before all its upstream work finishes.
  • The while remaining loop performs a topological sort. Each pass resolves pipelines whose dependencies are already scheduled. If no progress is made, a circular dependency exists.
  • This schedule deliberately ignores resource capacity. Multiple pipelines can pile into the same time slots, creating the resource spike you will visualize later.
  • Now replace your existing main() function and the if __name__ block with this updated version that also adds print_schedule() and calls the naive scheduler:
def print_schedule(schedule, title):
    # Display a formatted table of pipeline start/end times
    print(f"\n{'='*50}")
    print(f"  {title}")
    print(f"{'='*50}")
    print(f"  {'Pipeline':<20} {'Start':>6} {'End':>6} {'Resources':>10}")
    print(f"  {'-'*44}")
    for name, info in sorted(schedule.items(), key=lambda x: x[1]["start"]):
        end = info["start"] + info["duration"]
        print(f"  {name:<20} {info['start']:>6} {end:>6} {info['resource_demand']:>10}")


def main():
    pipelines, capacity, horizon = load_pipelines()

    print(f"Loaded {len(pipelines)} pipelines (capacity: {capacity}, horizon: {horizon})")
    print("\nPipelines:")
    for p in pipelines:
        deps = ", ".join(p["dependencies"]) if p["dependencies"] else "none"
        print(f"  {p['name']:<20} duration={p['duration']} demand={p['resource_demand']} deps=[{deps}]")

    # Run the naive (dependency-only) scheduler
    naive = build_naive_schedule(pipelines)
    naive_makespan = max(s["start"] + s["duration"] for s in naive.values())
    print_schedule(naive, f"NAIVE SCHEDULE (makespan={naive_makespan})")


if __name__ == "__main__":
    main()

What does this code do?

  • print_schedule() takes any schedule dictionary and renders it as a formatted table sorted by start time.
  • The updated main() now calls build_naive_schedule(), computes the makespan (when the last pipeline finishes), and prints the full naive schedule table.
  • Save optimizer.py.

Before you run this, what do you think the makespan of the naive schedule will be? Remember, pipelines with no dependencies all start at slot 0, and downstream pipelines start the moment their last dependency finishes.

  • Run the optimizer to see both the pipeline summary and the naive schedule:
python optimizer.py

You should see 8 pipelines listed with their properties, followed by the naive schedule table showing each pipeline's start slot, end slot, and resource demand. The makespan should be 12 time slots.

Why does the naive schedule create a problem?

Look at the first few rows of the naive schedule. Pipelines like raw_ingestion, user_events, and transaction_load all start at slot 0 because they have no dependencies. Their combined resource demand is 2 + 1 + 2 = 5, which already maxes out your cluster capacity.

In a real environment, this causes Spark executor starvation and job failures. The constraint solver you build in the next step will spread these pipelines across time slots to stay within capacity.

Seeing a NameError for build_naive_schedule?

  • Make sure build_naive_schedule() is defined above print_schedule() and main() in your file. Python reads top to bottom, so the function must exist before main() calls it.
  • Confirm you saved the file after adding the function.
  • help me fix a NameError in optimizer.py

✔️ Awesome, I've got everything!

Great. Double check you have saved both pipelines.json and optimizer.py.

ⓧ I'd like to double check the full code

Here is the full pipelines.json:

{
  "cluster_capacity": 5,
  "horizon": 40,
  "pipelines": [
    {"name": "raw_ingestion", "duration": 3, "resource_demand": 2, "dependencies": []},
    {"name": "user_events", "duration": 2, "resource_demand": 1, "dependencies": []},
    {"name": "transaction_load", "duration": 4, "resource_demand": 2, "dependencies": []},
    {"name": "data_transform", "duration": 3, "resource_demand": 2, "dependencies": ["raw_ingestion"]},
    {"name": "user_profiles", "duration": 2, "resource_demand": 1, "dependencies": ["user_events"]},
    {"name": "transaction_agg", "duration": 3, "resource_demand": 2, "dependencies": ["transaction_load"]},
    {"name": "ml_feature_store", "duration": 4, "resource_demand": 3, "dependencies": ["data_transform", "user_profiles"]},
    {"name": "dashboard_refresh", "duration": 2, "resource_demand": 1, "dependencies": ["transaction_agg", "ml_feature_store"]}
  ]
}

For optimizer.py, confirm these sections appear in order from top to bottom:

  • Imports (json, collections, cp_model).
  • load_pipelines() function that reads JSON, validates dependencies, and returns pipelines, capacity, and horizon.
  • build_naive_schedule() function that schedules everything ASAP respecting only dependency order.
  • print_schedule() function that renders a formatted table.
  • main() function that loads pipelines, prints the summary, runs the naive scheduler, and prints the naive schedule.
  • if __name__ == "__main__" guard calling main().

You have your pipeline data loaded and a naive baseline that proves the problem. Next up, you will build the constraint model that solves it.

Build the Constraint Model

You have 8 pipelines loading and printing correctly, along with a naive schedule that shows the resource-spike problem.

Now you need to translate your real-world scheduling rules into mathematical constraints that a solver can reason about. Without constraints, the solver has no rules to enforce. Dependencies, resource limits, and the "finish as fast as possible" goal all become formal constraints in this step.

In this step, get ready to:
  • Create solver variables that represent each pipeline's start and end time.
  • Add precedence and resource constraints so the solver respects your DAG dependencies and cluster capacity.
  • Minimize makespan and run the solver to get an optimal schedule.
Create variables and precedence constraints

Every pipeline needs three things the solver can manipulate: a start time, an end time, and an interval that ties them together. The solver searches through possible assignments for these variables until it finds one that satisfies every constraint.

  • In optimizer.py, find the build_naive_schedule() function. Add the following new function directly below it (above print_schedule):
def build_optimized_schedule(pipelines, capacity, horizon):
    model = cp_model.CpModel()

    # Store each pipeline's solver variables in a named tuple
    task_type = collections.namedtuple("task_type", "start end interval duration demand")
    all_tasks = {}
    all_intervals = []
    all_demands = []

    for p in pipelines:
        name = p["name"]
        duration = p["duration"]
        demand = p["resource_demand"]

        # Integer variables bounded by [0, horizon]
        start_var = model.new_int_var(0, horizon, f"start_{name}")
        end_var = model.new_int_var(0, horizon, f"end_{name}")
        interval_var = model.new_interval_var(
            start_var, duration, end_var, f"interval_{name}"
        )

        all_tasks[name] = task_type(
            start=start_var,
            end=end_var,
            interval=interval_var,
            duration=duration,
            demand=demand,
        )
        all_intervals.append(interval_var)
        all_demands.append(demand)

What does this code do?

Each pipeline gets a start_var and end_var integer variable that the solver can assign any value between 0 and the horizon (40 time slots).

The interval_var ties start, duration, and end together. This lets the solver reason about "pipeline X occupies time slots 3 through 6" as a single object. The all_intervals and all_demands lists collect everything the cumulative constraint needs later.

  • Save optimizer.py.
Add cumulative resource constraints

The naive schedule ignores resource limits entirely. Multiple pipelines can start at the same time even if their combined resource demand exceeds your cluster capacity of 5.

What is a cumulative constraint?

A cumulative constraint enforces that at any point in time, the sum of resource demands of all running intervals does not exceed a capacity limit.

Think of it as a bucket that holds 5 units. Each running pipeline drops its demand into the bucket. The constraint guarantees the bucket never overflows, no matter when pipelines start.

Now you will complete the build_optimized_schedule() function by adding precedence constraints, the cumulative resource constraint, the optimization objective, and the solver invocation.

  • In optimizer.py, add the following code directly after the all_demands.append(demand) line (still inside build_optimized_schedule, but outside the for loop):
    # Precedence: downstream can't start before upstream finishes
    for p in pipelines:
        for dep in p["dependencies"]:
            model.add(all_tasks[p["name"]].start >= all_tasks[dep].end)

    # Cumulative: total resource usage at any time <= capacity
    model.add_cumulative(all_intervals, all_demands, capacity)

    # Objective: minimize the time the last pipeline finishes
    makespan = model.new_int_var(0, horizon, "makespan")
    model.add_max_equality(
        makespan, [all_tasks[p["name"]].end for p in pipelines]
    )
    model.minimize(makespan)

    # Run the solver
    solver = cp_model.CpSolver()
    status = solver.solve(model)

    if status == cp_model.OPTIMAL or status == cp_model.FEASIBLE:
        schedule = {}
        for p in pipelines:
            name = p["name"]
            schedule[name] = {
                "start": solver.value(all_tasks[name].start),
                "duration": p["duration"],
                "resource_demand": p["resource_demand"],
            }
        return schedule, solver.objective_value, solver.status_name(status)
    else:
        raise RuntimeError(f"Solver failed with status: {solver.status_name(status)}")

What does this code do?

  • The precedence loop enforces that for every dependency A->B, pipeline B cannot start until pipeline A finishes.
  • model.add_cumulative() ensures that at no point in time does the total resource demand exceed 5 (your cluster capacity).
  • model.add_max_equality() sets the makespan variable to the latest end time across all pipelines. model.minimize(makespan) tells the solver to find the assignment that makes that latest finish as early as possible.
  • The solver returns OPTIMAL if it found the best possible schedule, or FEASIBLE if it found a valid schedule but cannot guarantee it is the best.
  • Save optimizer.py.
  • Verify the solver works by running this quick import test:
python -c "from optimizer import build_optimized_schedule, load_pipelines; p, c, h = load_pipelines(); s, m, st = build_optimized_schedule(p, c, h); print(f'Makespan: {int(m)}, Status: {st}')"

You should see output like Makespan: 12, Status: OPTIMAL. The exact makespan value depends on how the solver assigns slots, but the status should say OPTIMAL.

Seeing a SyntaxError or IndentationError?

  • Make sure the precedence loop and everything after it is indented at the same level as the variable creation loop (one level inside the function).
  • Check that the return statement and raise statement are inside the if/else block, indented two levels inside the function.

Help me debug my build_optimized_schedule function indentation

Set the objective and solve

The solver function works in isolation. Now you need to wire it into main() so it runs both schedules, compares them, and saves the results for visualization.

  • In optimizer.py, find the last line of main() which currently reads:
    print_schedule(naive, f"NAIVE SCHEDULE (makespan={naive_makespan})")
  • Add the following code directly below that line, still inside main():
    # Run the optimized schedule
    optimized, opt_makespan, status = build_optimized_schedule(pipelines, capacity, horizon)
    print_schedule(optimized, f"OPTIMIZED SCHEDULE (makespan={int(opt_makespan)}, status={status})")

    # Print improvement summary
    print(f"\n{'='*50}")
    print(f"  IMPROVEMENT SUMMARY")
    print(f"{'='*50}")
    print(f"  Naive makespan:     {naive_makespan} time slots")
    print(f"  Optimized makespan: {int(opt_makespan)} time slots")
    print(f"  Cluster capacity:   {capacity} concurrent slots")
    print(f"  Solver status:      {status}")

    # Save results for the visualization step
    results = {
        "naive": naive,
        "optimized": optimized,
        "capacity": capacity,
        "naive_makespan": naive_makespan,
        "optimized_makespan": int(opt_makespan),
    }
    with open("results.json", "w") as f:
        json.dump(results, f, indent=2)
    print("\n  Results saved to results.json")

What does this code do?

  • It calls your new build_optimized_schedule() function and prints the optimized pipeline assignments alongside the naive ones.
  • The improvement summary compares the two makespans so you can see the time saved at a glance.
  • Saving to results.json gives the visualization script (next step) the data it needs to render charts.
  • Save optimizer.py.

Before you run this, what do you think the solver's status will be? Will it find an optimal schedule, or just a feasible one?

  • Run the full optimizer by executing:
python optimizer.py

You should see both the naive and optimized schedules printed, with the solver reporting Status: OPTIMAL. The optimized makespan is shorter than the naive one, and the improvement summary shows the difference.

  • Confirm results.json was created by running:
ls

You should see results.json listed alongside optimizer.py, pipelines.json, and visualize.py.

Solver reporting INFEASIBLE?

  • Check that your pipelines.json has "horizon": 40. If the horizon is too small for the total pipeline durations, no valid schedule exists.
  • Make sure "cluster_capacity": 5 is set. A capacity of 1 would make it impossible to schedule pipelines with resource demands of 2 or 3.
  • Verify there are no circular dependencies in your JSON (e.g., A depends on B and B depends on A).

Help me debug why my CP-SAT solver is returning INFEASIBLE

Why CP-SAT instead of Airflow's built-in pools?

Airflow pools throttle concurrency reactively. They limit how many tasks run at once, but they do not plan ahead. If three heavy pipelines happen to land at the same time, one just waits.

CP-SAT is proactive. It considers all pipelines, all dependencies, and all time slots simultaneously to find the globally optimal assignment. It proves the schedule is optimal, not just "good enough."

✔️ Awesome, I've got everything!

Double check that you have saved optimizer.py and that results.json exists in your project folder.

ⓧ I'd like to double check the full code

Your optimizer.py is long at this point. Scroll through and confirm you have these sections in order:

  • Imports (json, collections, cp_model) at the top.
  • load_pipelines() function.
  • build_naive_schedule() function.
  • build_optimized_schedule() function (variables, precedence, cumulative, makespan, solver).
  • print_schedule() function.
  • main() function that runs both schedules, prints results, and saves results.json.
  • The if __name__ == "__main__": guard at the bottom.

Your optimizer now finds optimal start times for every pipeline while respecting both dependency order and resource limits. Next up, you will turn those numbers into a visual before/after comparison your boss can understand at a glance.

Visualize the Before/After

Your optimizer produces the right numbers, but numbers alone won't convince your boss to adopt a new scheduling approach.

A side-by-side Gantt chart with resource utilization bars makes the value of optimization immediately obvious without reading code. In this step, you will build the visualization layer that turns solver output into a presentation-ready PNG.

In this step, get ready to:
  • Build a Gantt chart renderer that draws pipelines as horizontal bars.
  • Add a resource utilization overlay showing capacity breaches.
  • Render a side-by-side comparison and save it as a PNG.
Load the solver results into visualize.py

The solver saved its output to results.json in the previous step. Your visualization script needs a function to read that file back in.

  • Open visualize.py in VS Code and add the imports and data-loading function:
import json
import matplotlib.pyplot as plt
import matplotlib.patches as mpatches
import numpy as np


def load_results(path="results.json"):
    with open(path) as f:
        return json.load(f)

What does this code do?

The imports bring in matplotlib for chart rendering and numpy for numerical helpers.

The load_results() function reads the JSON file that optimizer.py saved. It returns a dictionary containing both the naive and optimized schedules, capacity, and makespan values.

  • Save visualize.py.
  • Verify the import works by running this command:
python -c "from visualize import load_results; print(list(load_results().keys()))"

You should see the keys from your results file printed: ['naive', 'optimized', 'capacity', 'naive_makespan', 'optimized_makespan'].

Seeing a ModuleNotFoundError?

Make sure your virtual environment is still active (you should see (venv) in your terminal prompt). If not, run source venv/bin/activate again.

Also confirm you are running the command from inside the pipeline-optimizer folder where results.json exists.

Help me debug my visualize.py import error.

Build the Gantt chart renderer

A Gantt chart draws each pipeline as a horizontal bar positioned at its start time slot. This makes dependency order and parallelism instantly visible.

  • Add the render_schedule() function below load_results() in visualize.py:
def render_schedule(ax, schedule, title, capacity, colors):
    pipelines = sorted(schedule.items(), key=lambda x: x[1]["start"])
    y_labels = []

    for i, (name, info) in enumerate(pipelines):
        ax.barh(
            i,
            info["duration"],
            left=info["start"],
            height=0.6,
            color=colors.get(name, "#4A90D9"),
            edgecolor="black",
            linewidth=0.5,
        )
        y_labels.append(name)

    ax.set_yticks(range(len(y_labels)))
    ax.set_yticklabels(y_labels, fontsize=8)
    ax.set_xlabel("Time Slot")
    ax.set_title(title, fontweight="bold")
    ax.grid(axis="x", alpha=0.3)

What does this code do?

  • It sorts pipelines by start time so the chart reads top-to-bottom in execution order.
  • Each pipeline becomes a horizontal bar (ax.barh) starting at its assigned time slot, with width equal to its duration.
  • Color-coding comes from a color map passed in, so the same pipeline has the same color across both naive and optimized charts.
  • Save visualize.py.
Add the resource utilization overlay

The Gantt chart shows timing, but not whether you are overloading your cluster. A resource usage chart draws a filled area showing total demand at each time slot, with a red dashed line at the capacity limit.

  • Add the compute_resource_usage() helper function below render_schedule():
def compute_resource_usage(schedule, horizon):
    usage = [0] * horizon
    for name, info in schedule.items():
        for t in range(info["start"], info["start"] + info["duration"]):
            if t < horizon:
                usage[t] += info["resource_demand"]
    return usage

What does this code do?

For every time slot in the horizon, it sums up the resource demands of all pipelines that are actively running during that slot. The result is a list where each index represents a time slot and the value represents total cluster load at that moment.

  • Now add the render_resource_chart() function directly below compute_resource_usage():
def render_resource_chart(ax, schedule, capacity, horizon, title):
    usage = compute_resource_usage(schedule, horizon)
    x = range(len(usage))

    ax.fill_between(x, usage, alpha=0.4, color="#4A90D9")
    ax.plot(x, usage, color="#2C5F8A", linewidth=1.5)
    ax.axhline(y=capacity, color="red", linestyle="--", linewidth=1.5, label="Capacity limit")
    ax.set_xlabel("Time Slot")
    ax.set_ylabel("Resource Usage")
    ax.set_title(title, fontsize=9)
    ax.legend(fontsize=8)
    ax.set_ylim(0, capacity + 2)
    ax.grid(axis="y", alpha=0.3)

What does this code do?

  • It calls compute_resource_usage() to get the per-slot totals, then renders a filled area chart.
  • The red dashed line (ax.axhline) draws the cluster capacity limit. In the naive chart, you will see the blue area breach this line. In the optimized chart, it stays below.
  • Save visualize.py and confirm all functions import without errors:
python -c "from visualize import render_schedule, render_resource_chart; print('Resource chart ready')"

You should see Resource chart ready printed with no errors.

Seeing a SyntaxError?

Check that each function definition starts with def and ends with a proper colon. Also verify that the ax.barh() call in render_schedule() has matching parentheses across its multi-line arguments.

Help me fix a syntax error in my visualize.py file.

Render the side-by-side comparison and save

Now you need a main() function that loads the results, assigns colors, and arranges everything into a 2x2 figure: naive Gantt (top-left), optimized Gantt (top-right), naive resource usage (bottom-left), optimized resource usage (bottom-right).

  • Add the main() function and entry point at the bottom of visualize.py:
def main():
    results = load_results()
    naive = results["naive"]
    optimized = results["optimized"]
    capacity = results["capacity"]
    horizon = max(
        max(s["start"] + s["duration"] for s in naive.values()),
        max(s["start"] + s["duration"] for s in optimized.values()),
    ) + 2

    # Assign consistent colors across both charts
    all_names = list(naive.keys())
    cmap = plt.cm.get_cmap("tab10", len(all_names))
    colors = {name: cmap(i) for i, name in enumerate(all_names)}

    # Create 2x2 figure layout
    fig, axes = plt.subplots(2, 2, figsize=(14, 8))
    fig.suptitle(
        "Pipeline Schedule Optimizer: Before vs. After",
        fontsize=14,
        fontweight="bold",
    )

What does this code do?

  • It calculates the horizon dynamically from the actual schedule data (plus a small buffer) so the chart axes fit the content.
  • The tab10 colormap assigns a unique color to each pipeline name. Because both charts use the same color dictionary, the same pipeline is always the same color in naive and optimized views.
  • The 2x2 subplot grid (plt.subplots(2, 2)) gives you four panels for the four chart types.
  • Continue in the same main() function. Add the render calls and file save directly below the fig.suptitle() block:
    render_schedule(axes[0, 0], naive, f"Naive Schedule (makespan={results['naive_makespan']})", capacity, colors)
    render_schedule(axes[0, 1], optimized, f"Optimized Schedule (makespan={results['optimized_makespan']})", capacity, colors)
    render_resource_chart(axes[1, 0], naive, capacity, horizon, "Naive: Resource Usage Over Time")
    render_resource_chart(axes[1, 1], optimized, capacity, horizon, "Optimized: Resource Usage Over Time")

    plt.tight_layout()
    plt.savefig("schedule_comparison.png", dpi=150, bbox_inches="tight")
    print("Saved: schedule_comparison.png")
    plt.show()


if __name__ == "__main__":
    main()

What does this code do?

  • The four render calls fill each quadrant: top row is Gantt charts (naive left, optimized right), bottom row is resource usage charts.
  • The plt.tight_layout() call prevents labels from overlapping between subplots.
  • The figure is saved as a 150 DPI PNG before plt.show() opens it in a window. This gives you a file ready for a slide deck.
  • Save visualize.py.

Before you run this, what do you expect to see in the naive resource chart compared to the optimized one?

  • Run the visualization script:
python visualize.py

You should see Saved: schedule_comparison.png printed in your terminal. A matplotlib window opens showing your 2x2 chart.

In the naive resource chart (bottom-left), the blue area spikes above the red dashed capacity line. In the optimized chart (bottom-right), it stays below. That is the visual proof your boss needs.

Chart window not appearing?

Some environments (remote servers, WSL without display forwarding) cannot open a GUI window. The PNG file is still saved regardless. Confirm by running:

ls schedule_comparison.png

If the file exists, your visualization worked. Open it manually in your file browser or image viewer.

Help me get matplotlib to display charts on my system.

✔️ Awesome, I've got everything!

Great. Double check you have saved visualize.py and that schedule_comparison.png exists in your project folder.

ⓧ I'd like to double check the full code

Scroll up and confirm your visualize.py contains each of these sections in order:

  • Imports (json, matplotlib.pyplot, matplotlib.patches, numpy).
  • The load_results() function.
  • The compute_resource_usage() function.
  • The render_schedule() function.
  • The render_resource_chart() function.
  • The main() function with plt.subplots(2, 2), all four render calls, plt.savefig(), and plt.show().
  • The if __name__ == "__main__": entry point calling main().

You now have a visual artifact that proves your optimizer works. Next up, you will turn this optimized schedule into something your cluster can actually execute.

Secret mission

Generate Airflow DAG Output

Your chart proved the optimizer works. But your boss will ask: can we actually use this? In this secret mission, you will generate real Airflow-compatible cron expressions and a DAG Python file from the solver output, bridging the gap from proof-of-concept to production.

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, so there are no ongoing costs.

Resources you used:

  • Python virtual environment (venv/) with ortools and matplotlib installed.
  • Project folder (pipeline-optimizer/) containing Python scripts, JSON configs, and output files.

Keep everything running

No action needed. Choose this if you're still actively experimenting with the optimizer or want to tweak pipeline configurations and re-run the solver.

  • Your virtual environment stays active in the terminal session.
  • All project files remain in ~/Desktop/pipeline-optimizer.
  • You can re-run python optimizer.py or python visualize.py any time to regenerate schedules and charts.

Pause - I'll come back to this later

Deactivate the virtual environment to free up your terminal session. Your files and installed packages remain on disk for next time.

  • Deactivate the virtual environment by running:
deactivate
  • When you come back, navigate to ~/Desktop/pipeline-optimizer and reactivate with source venv/bin/activate (macOS/Linux) or venv\Scripts\activate (Windows).

Delete - I don't want to use this again

Remove the entire project folder including the virtual environment, all scripts, and generated output files.

  • Deactivate the virtual environment first by running:
deactivate
  • Delete the project folder by running:

macOS

cd ~/Desktop
rm -rf pipeline-optimizer

Windows

cd ~\Desktop
rmdir /s /q pipeline-optimizer
  • Confirm removal by running ls ~/Desktop (macOS) or dir ~\Desktop (Windows). The pipeline-optimizer folder should no longer appear.

Nice Work!

Nice Work!

You just built a production-ready pipeline schedule optimizer that proves constraint solvers can replace manual schedule tuning. Next time your team is fighting over cron schedules, you have the demo to back up the conversation.

You've learned how to:

  • Build a constraint programming model using OR-Tools CP-SAT that schedules pipelines while respecting DAG dependencies and cluster capacity limits.
  • Minimize total makespan by translating real-world rules into precedence constraints and cumulative resource constraints that prevent cluster overload.
  • Visualize the before/after improvement with Gantt charts and resource utilization overlays showing the naive spike vs. optimized spread.
  • Secret Mission: Generated Airflow-compatible cron expressions and a DAG file from the optimized schedule.

Ready to quiz yourself?