Building an End-to-End MLOps Pipeline
An end-to-end MLOps pipeline joins data validation, versioning, tracking, a registry, serving, monitoring and retraining into one loop. This guide runs that loop on real motorway traffic data: monitoring caught a summer surge, one retrained challenger was rejected, a second was promoted, and 2018 error fell 13%.
GloryTecks Team
Training & Placement Faculty
Last updated: 8 October 2026
Each MLOps practice makes sense on its own: validate data, version it, track experiments, register models, serve them, monitor them, retrain them. The value comes from joining them into one loop, where each step leaves something the next can rely on. This guide runs that loop end to end on real data, including the moments when it has to make decisions.
The data and the task
The UCI Metro Interstate Traffic Volume data record hourly westbound traffic on I-94 at a counting station between Minneapolis and St Paul, with weather and holidays, from 2012 to 2018, under CC BY 4.0 (UCI: Metro Interstate Traffic). The model predicts vehicles per hour from the hour, day, month, holiday flag and weather, as a traffic agency might for staffing or signage.
The loop
- Validate the raw data and reject impossible rows.
- Version the cleaned data with DVC, which tracks large data alongside code without storing it in Git (DVC: get started).
- Train and register with MLflow, recording the data version in every run.
- Serve whatever version the registry's champion alias points to.
- Monitor error month by month once true counts arrive.
- Respond with challengers that must beat the champion on data they never saw.
import os
import shutil
import stat
import subprocess
import sys
from contextlib import asynccontextmanager
import mlflow
import numpy as np
import pandas as pd
from fastapi import FastAPI
from fastapi.testclient import TestClient
from mlflow import MlflowClient
from sklearn.ensemble import HistGradientBoostingRegressor
os.environ["MLFLOW_RECORD_ENV_VARS_IN_MODEL_LOGGING"] = "false" # keep environment variable names out of model files
os.environ["MLFLOW_DISABLE_AGENT_HINT"] = "1"
def make_writable(func, path, _): # DVC's cache files are read-only on Windows
os.chmod(path, stat.S_IWRITE)
func(path)
for leftover in (".dvc", "data", "mlruns"): # start from a clean folder
if os.path.exists(leftover):
shutil.rmtree(leftover, onerror=make_writable)
if os.path.exists("pipeline.db"):
os.remove("pipeline.db")
# 1. Ingest and validate
raw = pd.read_csv("Metro_Interstate_Traffic_Volume.csv.gz", parse_dates=["date_time"]) # UCI, CC BY 4.0
checks = {"duplicate hours": raw.date_time.duplicated(), "temperature 0 K": raw.temp <= 200,
"rain over 100 mm/h": raw.rain_1h > 100}
print("1. Validate:", f"{len(raw):,} rows; rejected:", ", ".join(f"{name} {m.sum():,}" for name, m in checks.items()))
traffic = raw[~(checks["duplicate hours"] | checks["temperature 0 K"] | checks["rain over 100 mm/h"])].copy()
traffic["holiday"] = traffic.groupby(traffic.date_time.dt.normalize()).holiday.transform(lambda s: s.notna().any()).astype(int)
traffic["hour"], traffic["weekday"], traffic["month"] = traffic.date_time.dt.hour, traffic.date_time.dt.weekday, traffic.date_time.dt.month
traffic["temp_c"] = (traffic.temp - 273.15).round(2)
traffic["weather"] = traffic.weather_main.astype("category").cat.codes
FEATURES = ["hour", "weekday", "month", "holiday", "temp_c", "rain_1h", "snow_1h", "clouds_all", "weather"]
# 2. Version the cleaned data
os.makedirs("data")
traffic[["date_time"] + FEATURES + ["traffic_volume"]].to_csv("data/traffic.csv", index=False)
subprocess.run([sys.executable, "-m", "dvc", "init", "--no-scm", "-q"], check=True)
subprocess.run([sys.executable, "-m", "dvc", "add", "-q", "data/traffic.csv"], check=True)
pointer = open("data/traffic.csv.dvc").read()
print(f"2. Version: kept {len(traffic):,} rows; DVC pointer md5 {pointer.split('md5: ')[1].split()[0]}")
# 3. Train, track and register
mlflow.set_tracking_uri("sqlite:///pipeline.db")
mlflow.set_experiment("traffic-volume")
client = MlflowClient()
TRUSTED = ["sklearn.ensemble._hist_gradient_boosting.binning._BinMapper",
"sklearn.ensemble._hist_gradient_boosting.predictor.TreePredictor",
"sklearn._loss.link.IdentityLink", "sklearn._loss.link.Interval", "sklearn._loss.loss.HalfSquaredError",
"sklearn._loss._loss.CyHalfSquaredError"]
def between(start, end):
return traffic[(traffic.date_time >= start) & (traffic.date_time < end)]
def mae(model, rows):
return float(np.abs(model.predict(rows[FEATURES]) - rows.traffic_volume).mean())
def train_and_register(start, end, check_start, check_end):
model = HistGradientBoostingRegressor(random_state=0).fit(between(start, end)[FEATURES], between(start, end).traffic_volume)
with mlflow.start_run(run_name=f"train {start} to {end}"):
mlflow.log_params({"train_start": start, "train_end": end, "data_md5": pointer.split("md5: ")[1].split()[0]})
mlflow.log_metric("holdout_mae", mae(model, between(check_start, check_end)))
info = mlflow.sklearn.log_model(model, name="model", registered_model_name="traffic-volume",
input_example=between(start, end)[FEATURES].head(2), skops_trusted_types=TRUSTED)
return model, info.registered_model_version
v1, version = train_and_register("2016-01-01", "2016-11-01", "2016-11-01", "2017-01-01")
client.set_registered_model_alias("traffic-volume", "champion", version)
baseline = mae(v1, between("2016-11-01", "2017-01-01"))
print(f"3. Train: version {version} on Jan-Oct 2016, hold-out MAE {baseline:.0f} vehicles an hour; alias champion -> v{version}")
# 4. Serve whatever the registry calls champion
state = {}
@asynccontextmanager
async def lifespan(app):
state["model"] = mlflow.sklearn.load_model("models:/traffic-volume@champion")
state["version"] = client.get_model_version_by_alias("traffic-volume", "champion").version
yield
app = FastAPI(lifespan=lifespan)
@app.post("/predict")
def predict(rows: list[dict]):
return {"version": state["version"], "vehicles": [round(float(v)) for v in state["model"].predict(pd.DataFrame(rows)[FEATURES])]}
morning = between("2017-01-03 06:00", "2017-01-03 09:00")[FEATURES].to_dict("records")
with TestClient(app) as http:
print("4. Serve:", http.post("/predict", json=morning).json(), "| actual",
between("2017-01-03 06:00", "2017-01-03 09:00").traffic_volume.tolist())
# 5. Monitor 2017 month by month
print(f"5. Monitor 2017, alert when MAE > 1.25 x {baseline:.0f}:")
production = between("2017-01-01", "2018-01-01")
for month, rows in production.groupby(production.date_time.dt.strftime("%Y-%m")):
error = mae(v1, rows)
flag = "ALERT" if error > 1.25 * baseline else ""
print(f" {month} {error:4.0f} {flag}".rstrip())
if flag: # investigate: which days went most wrong?
daily = (v1.predict(rows[FEATURES]) - rows.traffic_volume).groupby(rows.date_time.dt.date).mean().nsmallest(3)
print(" biggest daily under-predictions:", ", ".join(f"{pd.Timestamp(d):%a %d %b} {e:+.0f}" for d, e in daily.items()))
# 6. Respond: challengers must beat the champion on a month they never saw
for name, (start, end, check_start, check_end) in {"A": ("2016-07-01", "2017-07-01", "2017-07-01", "2017-08-01"),
"B": ("2016-08-01", "2017-08-01", "2017-08-01", "2017-09-01")}.items():
challenger = HistGradientBoostingRegressor(random_state=0).fit(between(start, end)[FEATURES], between(start, end).traffic_volume)
new, old = mae(challenger, between(check_start, check_end)), mae(v1, between(check_start, check_end))
verdict = "promote" if new < 0.95 * old else "reject"
print(f"6. Challenger {name}, trained {start} to {end}: {new:.0f} vs champion {old:.0f} on {check_start[:7]} -> {verdict}")
if verdict == "promote":
_, version = train_and_register(start, end, check_start, check_end)
client.set_registered_model_alias("traffic-volume", "champion", version)
print(f" registered version {version}; alias champion -> v{version}")
with TestClient(app) as http: # a restart picks up the new champion
print("7. Serve after promotion:", http.post("/predict", json=morning).json())
new_champion = mlflow.sklearn.load_model("models:/traffic-volume@champion")
rows_2018 = between("2018-01-01", "2018-10-01")
print(f"8. Jan-Sep 2018 MAE: version 1 {mae(v1, rows_2018):.0f}, champion {mae(new_champion, rows_2018):.0f}")1. Validate: 48,204 rows; rejected: duplicate hours 7,629, temperature 0 K 10, rain over 100 mm/h 1
2. Version: kept 40,564 rows; DVC pointer md5 dd47e26555fe1d55294ca5f2940b4753
3. Train: version 1 on Jan-Oct 2016, hold-out MAE 357 vehicles an hour; alias champion -> v1
4. Serve: {'version': 1, 'vehicles': [5506, 6232, 5626]} | actual [5107, 6065, 5338]
5. Monitor 2017, alert when MAE > 1.25 x 357:
2017-01 275
2017-02 216
2017-03 210
2017-04 214
2017-05 190
2017-06 185
2017-07 513 ALERT
biggest daily under-predictions: Sat 22 Jul -887, Sat 15 Jul -785, Sun 16 Jul -783
2017-08 332
2017-09 363
2017-10 373
2017-11 412
2017-12 416
6. Challenger A, trained 2016-07-01 to 2017-07-01: 516 vs champion 513 on 2017-07 -> reject
6. Challenger B, trained 2016-08-01 to 2017-08-01: 315 vs champion 332 on 2017-08 -> promote
registered version 2; alias champion -> v2
7. Serve after promotion: {'version': 2, 'vehicles': [5235, 5912, 5409]}
8. Jan-Sep 2018 MAE: version 1 285, champion 248What happened
- Validation earned its place immediately. Of 48,204 raw rows, 7,629 repeated an hour already present, ten recorded a temperature of 0 kelvin and one claimed 9,831 mm of rain in an hour. Unchecked, those rows would have trained the model.
- The data version travelled with the model. Each MLflow run logged the DVC pointer's md5 hash, so any model can be traced back to the exact file it learned from.
- Serving followed the registry. The service loaded whatever the champion alias pointed to at start-up, using FastAPI's lifespan hook, which runs once before the app accepts requests (FastAPI: lifespan events). MLflow aliases are mutable, named references to a model version, which is what makes that work (MLflow: model registry).
- Monitoring caught a real change. Version 1, trained on early 2016, did better than its hold-out error through June 2017, then jumped to 513 vehicles an hour in July, above the alert line of 1.25 times 357. The pipeline then looked for the worst days: all three were weekends, Saturday 22 July, Saturday 15 July and Sunday 16 July, each under-predicted by between 783 and 887 vehicles an hour on average. Those weekends carried far more traffic than the 2016 data had taught the model to expect.
- The first response failed, and the gate stopped it. Challenger A, retrained on the twelve months to June, scored 516 on July against the champion's 513, so it was rejected. It had not seen the new pattern either.
- The second response worked. Once August's counts arrived, challenger B, trained up to the end of July, scored 315 on August against 332, clearing the required 5% improvement. It was registered as version 2, the champion alias moved, and the restarted service answered with version 2.
- The promotion held up. Over January to September 2018, the new champion's error was 248 vehicles an hour against version 1's 285, 13% lower.
Why the gate mattered
Without the gate, the pipeline would have shipped challenger A the moment the alert fired, replacing the model with one that was slightly worse. Retraining is a response, not a fix; the fix is a better model, and only an evaluation on unseen recent data can tell the two apart. For the reasoning behind gates and how reliable they are, see continuous training.
Where each stage goes deeper
- Data checks: testing machine learning code and data.
- Data versioning: data versioning with DVC.
- Registry and aliases: the model registry.
- Serving: serving ML models with FastAPI and Docker.
- Monitoring: monitoring ML models in production.
Taking it further
The script runs every stage in one process for clarity. In a real system, an orchestrator would schedule each stage, CI would test the code, a tracking server would replace the local database, and the service would run in a container behind a load balancer. The structure stays the same: versioned inputs, registered outputs, an alias as the deployment pointer, and gates between them.
Common mistakes
- Validating data only once, at the start of the project.
- Training without recording the data version.
- Serving a file path instead of a registry reference.
- Retraining on every alert without comparing against the champion.
- Judging a challenger on data it trained on.
What is an end-to-end MLOps pipeline?
What tools do I need for an end-to-end MLOps pipeline?
How do I deploy a new model version safely?
Can I build an MLOps pipeline without the cloud?
The End-to-End ML Pipeline is one of the projects in the MLOps course.
Sources
- Metro Interstate Traffic Volume dataset (UCI Machine Learning Repository) — UCI Machine Learning Repository
- Get Started with DVC (DVC documentation) — DVC
- ML Model Registry (MLflow documentation) — MLflow
- Lifespan Events (FastAPI documentation) — FastAPI
GloryTecks Team
Training & Placement Faculty
The GloryTecks faculty team mentors learners across Data Science, AI and Data Engineering, with a focus on real-time projects and placement readiness in Hyderabad.
Want to master MLOps?
Join GloryTecks’ expert-led MLOps training in Hyderabad with real-time projects and 100% placement support.
Related articles
MLOps Roadmap 2026: Skills, Tools and Career Path
MLOps sits between data science, software engineering and operations, which makes it hard to know where to start. This roadmap orders the skills from foundations to production operations, gives each stage a project that proves the skill, and explains what the MLOps engineer role actually involves.
MLflow Tutorial: Tracking Experiments and Models
MLflow records what produced each model so you never have to remember. This tutorial trains three models on real bike-sharing data, logs every run, compares them by querying the tracking store, registers the best one, points a champion alias at it and loads it back by that alias, all on a laptop.
Docker for Machine Learning: A Practical Guide
How to containerise a machine-learning project so it runs the same everywhere: pinning the exact library versions a saved model depends on, a Dockerfile ordered for fast rebuilds, keeping data out of images, multi-stage builds, non-root users and GPU access.