Automating the training
We got to a pretty good place last time, we had a model that we could train, evaluate and serve with just a couple of commands but we've got a few big problems that we need to tackle left over before we can really use this in a production environment.
MLOps is different from traditional DevOps in that we're aiming at a moving target.
- Our model works great today, but what about a year from now?
- How do we manage doing that training? Does someone have to remember to come into the project and run it?
- How long is long enough to capture the benefit from training compared to just burning processing time for nothing?
- What happens if the drivers, constructors and circuits change enough such that our model doesn't work well?
- What if the relationships change e.g. battery tech changes the game sufficiently that our features don't capture it well anymore?
Fortunately, to orchestrate this we have Prefect.
Setting up Prefect
First up, what is it? It's an orchestrator.
It's the modern equivalent of AirFlow which makes directed acyclic graph's. Yeah my eyes glazed over too, basically it means we specify what we want and it figures out for us how to plug it all together to give us what we need.
The good thing is that unlike AirFlow it's all just python decorators on functions so it's super quick to setup. What we need to do is just add it to our docker compose to get it running.
prefect:
image: prefecthq/prefect:3-latest
container_name: f1-prefect
ports:
- "127.0.0.1:4200:4200"
depends_on:
lakefs:
condition: service_healthy
mlflow:
condition: service_started
volumes:
- prefect-data:/root/.prefect
command: prefect server start --host 0.0.0.0
healthcheck:
test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:4200/api/health')"]
interval: 5s
timeout: 5s
retries: 10
...
volumes:
...
prefect-data:That gives us the prefect server but without something to do the work, it's basically an empty shell.
Work work.
Traditionally, we would do the following:
- Create a work pool
- Create a deployment
- Register a worker
- Have the worker pick up the job
Having all of these separate steps makes a lot of sense if you want to manage multiple flows across multiple machines, use different infrastructure for different deployments, scale specific workers etc. But for our purposes it's overkill so what we'll do instead is go with the simpler runner approach.
if __name__ == "__main__":
asyncio.run(f1_pipeline.serve(
name="scheduled",
cron=settings.prefect_cron_schedule
))The .serve here is the simpler syntax, this is doing three different things under the hood:
- Registers the deployment with Prefect, this allows it to have a schedule, and allows for manual triggering
- Polls for runs that are due
- Executes the runs that are due within the same process
So, how do we set this up? It's super easy, we just create a function with the following decorator. This creates a top level flow, and then underneath that we have tasks which are like the individual jobs.
@flow(name=settings.prefect_flow_name, log_prints=True)
async def f1_pipeline():
# Some code here
run_training("We put our reason here")
@task
async def run_training(reason: str):
# Calls into our existing pipeline
We don't want to have to run this manually ourselves, we want this sat somewhere and running so we write a Dockerfile for the service. This is important because it's going to be doing the training and fetching of the data later on, so we need to bundle more than the serving portion.
FROM python:3.12-slim
WORKDIR /app
RUN apt-get update && apt-get install -y libgomp1 && rm -rf /var/lib/apt/lists/*
COPY pyproject.toml .
COPY src/ src/
COPY ingest/ ingest/
RUN pip install -e ".[train,data,flows]"
CMD ["python", "src/f1_predictor/flows/pipeline_flow.py"]Finally, we just need to wire this into the docker compose file, this registers it as an agent effectively with Prefect ready to pick up training jobs!
prefect-worker:
build:
context: .
dockerfile: Dockerfile.worker
container_name: f1-prefect-worker
depends_on:
prefect:
condition: service_healthy
environment:
- PREFECT_API_URL=http://f1-prefect:4200/api
env_file: .env.docker
command: python src/f1_predictor/flows/pipeline_flow.py
Now we've got our worker ready to go
Fetching fresh data
So every X days or however long, we trigger retraining. But, right now this doesn't really help us at all...
Why? Well, because we're just retraining on the same data over and over again at the moment because we're still using the ingested .csv files from the kaggle dataset. So, what we need to do is the following to make it worthwhile:
- Fetch the latest data from the tip of the branch in LakeFS
- Pull the latest data from the Jolpica API
- Transform the data into our existing structure
- Commit it to the staging branch
- Clean & validate the data
- If we pass, merge the data from staging into main
The fetching data bit we just grab the data from LakeFS, easy peasy:
class LakeFSRepository():
def __init__(self):
self._clt = Client(
host=settings.lakefs_host,
username=settings.lakefs_installation_access_key_id,
password=settings.lakefs_installation_secret_access_key,
)
self._repo = lakefs.Repository(settings.lakefs_repo, client=self._clt)
self._branch = self._repo.branch("main")
def _read_csv(self, filepath: str, **kwargs) -> pd.DataFrame:
with self._branch.object(filepath).reader(mode="rb") as f:
return pd.read_csv(io.BytesIO(f.read()), **kwargs)
...
def load_races(self) -> pd.DataFrame:
return self._read_csv("raw/races.csv", na_values="\\N", parse_dates=["date", "quali_date"])But, we don't want to fetch all the data every time, we'd be throwing away a huge portion of it as really there should only be one extra race each weekend. So what we can do is just find the last race that occured, and bump the count to get the next year. Don't worry, I know you're thinking "What if it's the last race in the season?" If we don't find the race we try setting the round to 1 and bumping the year instead so that's handled.
def get_data_from_api(lakefsData: F1Data) -> F1Data:
f1_api_repo = JolpicaRepository()
last_row = pd.merge(
left=lakefsData.results,
right=lakefsData.races,
how="left",
on="raceId"
).sort_values(["year", "round"]).iloc[-1]
latest_year = last_row["year"]
latest_round = last_row["round"] + 1
latest_year, latest_round, race_result_data = f1_api_repo.get_next_race_results(latest_year, latest_round)The data from the API doesn't look exactly like the data we had before, so it needs a little extra cleaning to get it into the same format and we need to rename some of the columns but this is pretty easy to do. After that we need to reconcile the new data with the old, because races, constructors and circuits are all categorical we look at the existing data and see if any new ones have been added and create new id's if there are. Then we just append in the new race data.
Finally, write it back to LakeFS on the staging branch
staging_branch = lakefsRepo.create_branch("staging")
write_to_lakefs(updated_data, staging_branch)
print("Written to staging branch")Why staging? Why not directly onto main? We haven't cleaned it yet.
If the data isn't clean we'll want to have a store of it somewhere that we can inspect and do a little EDA to understand not only why did it fail our validation but to what extent it failed validation as well. Did something strange happen that we didn't forsee? Or is the API having issues and returning strange data? Having all of this on an unmerged branch makes it easier to deal with.
try:
merged_data = build_race_frame(
races=updated_data.races,
circuits=updated_data.circuits,
constructors=updated_data.constructors,
drivers=updated_data.drivers,
results=updated_data.results,
statuses=updated_data.statuses
)
validate.check_schema(clean.clean_data(merged_data, min_year=1990))
except Exception as e:
print(f"Validation failed — data remains on staging branch for inspection: {e}")
return
lakefsRepo.merge_into(source="staging", target="main", message=f"Merge validated race data to main: {latest_year} round {latest_round}")Finally, we can build our full race frame again, run the validation and merge the staging branch into main if it passes the validation. Now we've got fresh and clean data!
Putting it on a schedule
Scheduling is pretty easy with the Prefect setup that we have going on, you can do more complex setups but I think for now it's worth to setup a CRON schema for say, 6PM Sunday afternoon. That way regardless of where the race takes place in the world we can be sure that there is fresh data to fetch every week, if there hasn't been a race (for whatever reason) our ingestion pipeline will just quit out early so it's no problem.
prefect_cron_schedule: str = "0 18 * * 0"
So now we've got our model training every race weekend, but how do we use the result?
How we promote & why
Before we talk about how we promote models, we need to have a quick chat about training, evaluation, and test sets.
When we were doing feature engineering it made much more sense to have a held out test set, we needed to make sure that we had a pure base in order to reference our progress against in order to make sure that we didn't leak data through but could also evaluate the performance of the features in a pure way.
The problem is, that for a real-time predictive model like this one - it's simply a luxury we can't afford.
What happens instead that the future becomes our test set, we need to include all of the data up to the latest point because we need all of the training data that we can possibly get because we operate on rolling features, the most recent ones are the most valuable data we have, to hold them back is to operate on stale data.
Our test set is still there, it's just coming one race at a time.
This also changes the relationship with what the validation metric tells us. If ROC-AUC shfits after training is it because the models features have become less predictive, such as rolling rates no longer accurately capture driver form well, or if the underlying relationships have changed such as a regulation change reshuffling the competitive order in a way we can't see (The battery world championship...).
We keep the rolling window cross-validiation during training, it trains on a set of years and validates that concept on the next held out year so it operates more as a sanity check that then model generalizes over time, not necessarily as a held out test set.
client.set_registered_model_alias(
name=settings.mlflow_experiment_name,
alias="champion",
version=version.version
)
print(f"Champion alias promoted to version {version.version}")These are no longer concepts that we can manage purely at the training and evaluation stage, it becomes the responsibility of a secondary system that we'll talk about shortly, but the answer to our conundrum for now is always promote the latest.
Using prefect variables
This is very much a nice to have, but, moving the variables from the .env files and environment variables into prefects managment means that:
- We've got a nice UI to track them
- We don't have to touch the code at all to change parameters in the system
prefect-setup:
image: python:3.12-slim
depends_on:
prefect:
condition: service_healthy
restart: on-failure
env_file: .env.docker
entrypoint:
- python
- -c
- |
import urllib.request, urllib.error, json, os, sys
def create_var(name, value):
data = json.dumps({"name": name, "value": json.loads(value)}).encode()
req = urllib.request.Request(
"http://f1-prefect:4200/api/variables/",
data=data,
headers={"Content-Type": "application/json"},
method="POST"
)
try:
urllib.request.urlopen(req)
print(f"{name}: ok")
except urllib.error.HTTPError as e:
if e.code == 409:
print(f"{name}: already exists")
else:
print(f"{name}: failed with status {e.code}")
sys.exit(1)
create_var("lgbm_hyperparameters", os.environ["LGBM_HYPERPARAMETERS"])
create_var("retraining_config", os.environ["RETRAINING_CONFIG"])We keep the default values in the config value, and seed them into prefect just so that nobody has to do the initial setup. But it means then that going forward to can change the parameters there and they'll pull through:
The only extra bit of work we need to do is to modify the training flow such that it loads the variable values from Prefect and passes them down, from that point on everything else is the same as far as MLFlow and the training are concerned. So if we train a model, the model gets stamped with those variables that were used to train it
@task
async def run_training(reason: str):
lgbm_params = parse_variable(await Variable.get("lgbm_hyperparameters"))
print(f"Retraining triggered: {reason} - {lgbm_params}")
train_pipeline(lgbm_params)It's a small thing but it's important, Why? Anything an end user should want to do with our system should not require them to change the source code of the system.
You don't pop the hood of your car to connect wires when you want to start it, you use the key.
Drifting
Firstly, What is drift?
There are a few different types of drift that affect us really, these are:
- Data drift: The distribution of the inputs have shifted meaningfully, new drivers, new constructors, regulation changes that modify the competitiveness. The model hasn't seen these entities or distributions before so our quality drops.
- Concept drift: The inputs look the same, but the relationships change. The ground effect era didn't change anything in terms of features we could see before we added regulation era, the data is the same but something conceptually changes that we can't see that makes our predictions worse.
- Data quality drift: The inputs are more poorly formed and or missing than they were before, which obviously has an impact downstream
As someone personally from a DevOps background, it's worth thinking about this as rather than just maintaining the software to keep it alive, accurate and up to date. We also need to maintain the data that powers it too. We're now constantly chasing a moving target, even when we stay still the world moves.
So the tool that is going to help us to manage all of this is Evidently.
Setting up Evidently
We're going to be using the open source version and hosting it ourselves. However, it's not really built nicely to be hosted locally because it's intended to be run as a python library instead. That's not a problem however, what we do is we wrap it in a Dockerfile
# Dockerfile.evidently
FROM python:3.12-slim
RUN pip install evidently==0.6.7
CMD ["evidently", "ui", "--workspace", "/evidently/workspace", "--host", "0.0.0.0", "--port", "8000"]We could use a raw python image and do the installation each time, but this is quicker if you need to remove and restart the image a lot so it makes more sense to bake it all in. Once that is sorted we just add it to our docker compose
evidently:
build:
context: .
dockerfile: Dockerfile.evidently
container_name: f1-evidently
ports:
- "127.0.0.1:8001:8000"
volumes:
- evidently-data:/evidently/workspace
env_file: .env.docker
restart: unless-stopped
volumes:
...
evidently-data:Checking the drift
Okay, so we've got the infrastructure all set up. How do we hook this in?
The good news is that because we've already got the flow in prefect, and Evidently provides a lot of the reports we need batteries included.
DRIFT_EXCLUDE = ["year", "round", "driver_experience", "grid_size", "regulation_era"]
DRIFT_FEATURES = [f for f in MODEL_FEATURES if f not in DRIFT_EXCLUDE]
REPORT_FEATURES = DRIFT_FEATURES + ["podiumFinish"]
ALL_CATEGORICALS = ["driverId", "constructorId", "circuitId", "regulation_era", "is_home_race"]
@task
def check_drift(current_df: pd.DataFrame, reference_df: pd.DataFrame) -> bool:
column_mapping = ColumnMapping()
column_mapping.target = "podiumFinish"
column_mapping.categorical_features = [f for f in ALL_CATEGORICALS if f in DRIFT_FEATURES]
column_mapping.numerical_features = [
f for f in DRIFT_FEATURES
if f not in column_mapping.categorical_features
]
report = Report(metrics=[
DataDriftPreset(),
DataQualityPreset(),
TargetDriftPreset(),
])
report.run(
reference_data=reference_df[REPORT_FEATURES],
current_data=current_df[REPORT_FEATURES],
column_mapping=column_mapping
)
workspace = RemoteWorkspace(settings.evidently_workspace_url)
project = get_or_create_project(workspace, settings.evidently_project_name)
workspace.add_report(project.id, report)
result = report.as_dict()
return result["metrics"][0]["result"]["dataset_drift"]We need to tell evidently which features are categorical, also which features to include and which to include. Remember, the dataframe we pass in here has everything not just our engineered features so this is important. We also exclude a few features as well, there isn't much point looking for drift in year, round or driver_experience because they don't tell us anything meaningful about the data.
The type of test used is based on the type of feature that is being assessed, categorical, numerical and threshold feature all get different approaches which is why it was so important to tag them before.
In specifics
"See this moment! ... From this gateway moment a long eternal lane stretches backward: behind us lies an eternity." — Friedrich Nietzsche
Our drift report has a big flaw currently, the past is significantly more dominant than the present.
Currently, the drift report compares the latest races to all of the previous data that the model was trained upon, the difficulty with this is that no matter how much the newest race moves, there will never be sufficient enough drift to trigger the threshold.
What we do is to store a configurable amount of last 72 races as a .parquet file against the trained model like so:
def log_model_artifacts(model, X, y, data, run_name, last_n_races=72):
mlflow.log_params(model.get_params())
last_n_race_ids = (
data[["raceId", "year", "round"]]
.drop_duplicates()
.sort_values(["year", "round"])
.tail(last_n_races)["raceId"]
)
recent_mask = data["raceId"].isin(last_n_race_ids)
reference = X[recent_mask].copy()
reference["podiumFinish"] = y[recent_mask].values
with tempfile.TemporaryDirectory() as tmp_dir:
tmp_path = os.path.join(tmp_dir, "training_reference.parquet")
reference.to_parquet(tmp_path, index=False)
mlflow.log_artifact(tmp_path)
This then allows us to fetch that reference set against the current set, we build up the current set by taking say the last 12 races including the new data
@task
def build_current_frame(full_df: pd.DataFrame) -> pd.DataFrame:
"""
Slice the last 12 races from the full frame to use as the current
distribution for drift detection. Uses raceId to avoid assumptions
about race cadence or calendar.
"""
last_12_race_ids = (
full_df[["raceId", "year", "round"]]
.drop_duplicates()
.sort_values(["year", "round"])
.tail(12)["raceId"]
)
return full_df[full_df["raceId"].isin(last_12_race_ids)]
Comparing these two in the drift report means then that the past does not continuously expand meaning that drift becomes less visible over time, and we're looking at a big enough chunk of current races for it to be a significant amount of drift.
When to train
Knowing when to train is just as, if not more important than knowing what to train. How we set that threshold has two possible outcomes
- We don't train often enough: We miss model performance improvements in terms of tackling drift.
- We train too much: We spend money and water training models with little to no benefit.
Fortunately, even though this problem is a difficult one to get into a sweet spot for, our drift reports can help us a lot here. We're going to set two thresholds, either our dataset has drifted significantly or it has been X number of races.
drift_detected = check_drift(current_df, reference_df)
force_retrain = await should_force_retrain(full_df)
if drift_detected and force_retrain:
await run_training("drift detected and 12-race threshold reached")
elif drift_detected:
await run_training("drift detected")
elif force_retrain:
await run_training("12-race threshold reached")
else:
print("No retraining required")
Why the upper limit of 12 races? It just keeps the model operating on fresh data even if it hasn't necessarily drifted, it gives us a nice ceiling to work with to ensure that our model stays relevant.
This combined with the guards that we have around the fetching of the data mean that we can actually run the model much more frequently and unless both new race data is available, and either the data has drifted or it has been more than 12 races, we will avoid training a new model. meaning we can actually increase the frequency of the CRON job if we wanted to to make sure we don't miss anything. As nothing will be trained until it's necessary.
The post race interview.
Five parts, one complete system. Let's take stock of what we actually built and why the decisions we made along the way matter.
What we built
A defensible baseline. We rejected qualifying position as a baseline before we trained a single model because it leaks same-weekend signal. The rolling 5-race podium rate is the honest benchmark, everything has to beat that or it isn't worth shipping.
A model that knows what it doesn't know. LightGBM with rolling window cross-validation, Brier score as the primary metric. Not just "is this driver on the podium" but "how confident are we, and is that confidence well calibrated".
A data layer with a memory. LakeFS gives every training run a commit SHA. You can always answer the question "what data produced this model", which sounds obvious until you've worked on a system that can't.
A serving layer that doesn't need the race to have happened. The FastAPI server extrapolates features for future races from the most recent known data. It's an approximation, but it's an honest one.
A pipeline that runs without us. Prefect ingests new race data, validates it through a staging branch before it touches main, and makes a principled decision about whether to retrain. Nobody has to remember to do anything.
A monitoring layer that watches for the world changing. Evidently compares the last 12 races against a fixed reference window from the champion model's training run. The past doesn't expand, the signal doesn't get buried.
Decisions worth calling out
Rolling window CV, not a held-out test set. In production you can't afford to hold back the most recent data, those races are the most predictive ones we have. The future is the test set, it just arrives one race at a time.
Always promote the latest model. Gating promotion on a metric comparison sounds rigorous but it isn't. If the world has changed, a better metric against stale data isn't evidence of a better model. Drift monitoring is the quality gate, not a post-hoc number.
Two retraining guards, not one. Drift alone can miss gradual degradation that never crosses the statistical threshold. A race count ceiling alone could trigger unnecessary retrains when the distribution is perfectly stable. Together they cover the cases the other misses.
Staging before main. If new data fails validation you want to inspect it, not lose it. An unmerged staging branch gives you somewhere to do that EDA without it polluting the training data.
Prefect Variables for config. Anything a user should be able to change about the system shouldn't require them to touch the source code. Hyperparameters and retraining thresholds live in the UI, not in a file you have to redeploy to change.
What's left
There's always more. The serve layer doesn't reload automatically when a new champion is promoted, that's a gap worth closing if this were going to real production. SHAP values would tell us not just whether the model is drifting but which features are driving it. Optuna found our hyperparameters once; with the pipeline automated there's an argument for periodic re-tuning as the data grows.
But as a demonstration of what a production MLOps system looks like end to end from a blank notebook to automated retraining with drift monitoring. I think we got there. Lights out.
