opti-pipe — AI Data Pipeline Optimizer
The problem
Manual tuning is a tax every data team pays
Data engineers waste hours guessing at executor memory, batch sizes, and cluster<br>size — and get it wrong in both directions.
Runaway cloud costs
Over-provisioned clusters and executor memory sit idle, burning budget on<br>capacity nobody is using — and nobody notices until the bill arrives.
Failures & latency spikes
Under-provisioned memory and unoptimized parallelism cause OOM kills and<br>queue delay — usually discovered in production, at the worst time.
What it watches
Four vectors, pulled from your actual runs
opti-pipe reads config files alongside execution telemetry — no need to<br>eyeball dashboards across four different tools.
Latency<br>Execution duration
Stage/job completion time and queue delay, tracked per run to catch<br>bottlenecks before they compound.
Throughput<br>Records / sec
Processing rate derived per run, so scaling issues show up before a<br>pipeline falls behind its SLA.
Resource utilization<br>CPU, heap, GC pauses
Executor heap usage, allocated vs. actual CPU, and GC pause time — the<br>signals that separate "wasteful" from "about to OOM."
Cloud cost<br>Cost per run
Instance type, cluster size, and schedule combine into a monthly cost<br>estimate — and a savings number for every fix.
How it works
$ opti-pipe check spark-daily-etl-orders<br># pipeline already ran on schedule - nothing new to trigger here<br># comparing last 5 runs against the live config...
rule 01 fired: spark.executor.memory 16g -> 13g<br>$15.09/mo savings, +0.4% latency
# diff, rationale, and impact are all above - nothing has changed yet<br>$ _ waiting for your approval
I don't trust an agent to touch a production Spark config unattended, so it doesn't.<br>You approve every change in the dashboard below, or nothing happens.
Pipeline health at a glance — cost, savings, and active recommendations per pipeline.
Every recommendation shows the diff, the rationale, and the estimated impact — before you click Approve.<br>The +$34.06/mo and +$45.41/mo above are real, computed live from the bundled sample pipeline —<br>rerun the same file yourself and you'll get the same numbers. Fixing a real OOM risk isn't free, and it says so upfront instead of hiding it.
# seven rules, and you can read every one of them - analyzer.py, no model in between
$ grep -A1 "^def _rule" analyzer.py
def _rule_overprovisioned_memory(...) ## 01 cost<br>low heap, no OOM history -> shrink spark.executor.memory
def _rule_oom_or_gc_pressure(...) ## 02 reliability<br>OOM errors or high GC pause -> grow executor memory before it fails again
def _rule_shuffle_partitions(...) ## 03 latency<br>too many tiny tasks, or too few slow ones -> resize spark.sql.shuffle.partitions
def _rule_overprovisioned_cluster(...) ## 04 cost<br>low CPU, no queue delay -> shrink executor instance count
def _rule_queue_delay(...) ## 05 latency<br>jobs waiting on capacity -> grow executor instance count
def _rule_dbt_threads(...) ## 06 latency<br>low threads, slow runs -> raise DAG parallelism
def _rule_dbt_tasks_per_thread(...) ## 07 latency<br>too many models/tests per thread, even if not slow yet -> raise threads early
Rule 02 is the one I actually needed. Some Spark job would randomly OOM, I'd lose an<br>evening staring at heap and GC numbers trying to work out what changed, and eventually I got tired of<br>doing it by hand and wrote this instead. If it's useful to you too, good — if not, tell me why, that's<br>worth more to me than a compliment.<br>— I'm the only person who works on this
No auto-apply. Some optimization tools have an "autopilot" mode that<br>applies changes to your cluster after every run. I didn't build one — there's<br>no switch to turn it on. GET /recommendations recomputes live from<br>the current config and metrics on every call, so there's no hidden state that<br>could drift from what you last approved.
Optional AI layer
A second opinion, on demand
The seven rules above stay free, deterministic, and fully unit-tested — that<br>doesn't change. On top of them, an optional "Ask AI" button sends your<br>metrics and the rules' own output to a real Claude model for a plain-English<br>critique: the same natural-language second-opinion idea larger platforms<br>charge enterprise pricing for, here as one button with your own API key.
Summary<br>Plain-English health check
A few sentences on this pipeline's overall health, written from its<br>actual run metrics.
Critique<br>Checks the rules' own work
Does the deterministic engine's recommendation actually look right given<br>the numbers? The model is explicitly asked to disagree if it should.
Observations<br>What seven thresholds miss
Trends across runs, outliers, correlations between metrics — patterns no<br>fixed threshold rule is built to catch.
This part costs real money per click, so I made it optional instead of<br>automatic. Each click is a billed Claude API call, using your own<br>Anthropic API key (a claude.ai subscription doesn't cover API access).<br>Nothing else on this page needs one, and...