Building an End-to-End MLOps Pipeline for Bulldozer Price Prediction with Prefect and CometML

Machine learning models are only as valuable as your ability to deploy them reliably into production and monitor their performance over time. That‘s where MLOps comes in – applying DevOps best practices to streamline the entire machine learning lifecycle.

In this in-depth tutorial, we‘ll walk through building an end-to-end MLOps pipeline for predicting bulldozer sales prices using two powerful open source tools: Prefect for workflow orchestration and CometML for experiment tracking and model monitoring. By the end, you‘ll have a template you can adapt for your own ML use cases to boost productivity and collaboration.

Why Prefect and CometML for MLOps?

Prefect is a popular open-source tool for building, running, and monitoring robust data pipelines. It provides a clean Python framework for defining complex workflows as a graph of tasks, making it easy to parameterize, schedule, and observe your pipelines. Prefect also integrates well with a variety of orchestration platforms like Kubernetes and Apache Airflow.

CometML is an MLOps platform that focuses on experiment management, model monitoring, and collaboration. With CometML, you can easily log and compare experiments, track datasets and code changes, monitor models in production, and share insights with your team. It‘s a great complement to Prefect for the model development and monitoring aspects of MLOps.

The Bulldozer Price Prediction Pipeline

To demonstrate these tools in action, we‘ll build a realistic pipeline to predict the sale price of bulldozers based on historical auction data. The pipeline will include the following steps:

1. Ingest: Load the raw bulldozer auction data from a CSV file
2. Clean: Preprocess the data, handling missing values and converting datatypes
3. Split: Partition the data into train and test sets
4. Train: Fit a LightGBM model on the training set, tracking hyperparameters and metrics in CometML
5. Evaluate: Measure the model‘s performance on the test set and log to CometML
6. Monitor: Set up monitoring in CometML to track drift and integrity over time

We‘ll implement each step as a Prefect task and connect them into a scheduled workflow. Let‘s get started!

Step 1: Ingest

First, we need to load our raw data. We‘ll assume it‘s in a local CSV file, but this could easily be a database connection or cloud storage bucket. In ingest.py:

from prefect import task
import pandas as pd

@task
def load_data(path):
    df = pd.read_csv(path)
    return df

This simple Prefect task loads the auction data CSV into a Pandas DataFrame. The @task decorator lets Prefect track this function as a node in our workflow graph.

Step 2: Clean

Real-world data is messy, so some preprocessing is usually needed before training a model. We‘ll impute missing values and encode categorical variables in clean.py:

from prefect import task

@task
def preprocess_data(df):
    # Handle missing values
    df.fillna(0, inplace=True)

    # Convert category fields to integers
    cat_cols = ["UsageBand", "ProductGroup", "ProductGroupDesc", "Drive_System", 
        "Enclosure", "Forks", "Pad_Type", "Ride_Control", "Stick", "Transmission", 
        "Turbocharged", "Blade_Extension", "Blade_Width", "Enclosure_Type",
        "Engine_Horsepower", "Pushblock", "Ripper", "Scarifier", "Tip_Control"]

    for col in cat_cols:
        df[col] = pd.Categorical(df[col]).codes

    return df  

Step 3: Split

With our data cleaned, we can split it into training and test sets. We‘ll put this logic in split.py:

from prefect import task
from sklearn.model_selection import train_test_split

@task
def split_data(df, target_col, test_size=0.2):
    X = df.drop(columns=[target_col])
    y = df[target_col]

    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=test_size, random_state=42)

    return X_train, X_test, y_train, y_test

This uses scikit-learn‘s train_test_split function to partition our feature matrix X and target vector y into train and test subsets.

Step 4: Train

Now for the exciting part – training our price prediction model! We‘ll use the LightGBM library, which is known for its speed and performance on tabular data. The training logic goes in train.py:

from prefect import task
import lightgbm as lgb
from comet_ml import Experiment

@task 
def train_model(X_train, y_train):
    experiment = Experiment(project_name="bulldozer-prices")

    dtrain = lgb.Dataset(X_train, label=y_train)

    param = {"max_depth": 5, "learning_rate": 0.1, "n_estimators": 200}
    experiment.log_parameters(param)

    model = lgb.train(param, dtrain, 50)

    experiment.log_model("model", model)
    experiment.end()

    return model

Here we initialize a CometML Experiment to track our training run. We log the model hyperparameters as well as the model object itself for future reference. Calling experiment.end() ensures all metrics are flushed before exiting.

Step 5: Evaluate

With our model trained, let‘s see how it performs on the held-out test set. The evaluation code belongs in evaluate.py:

from prefect import task
from sklearn.metrics import mean_squared_error, r2_score
from comet_ml import Experiment
import numpy as np

@task
def evaluate_model(model, X_test, y_test):
    experiment = Experiment(project_name="bulldozer-prices")

    y_pred = model.predict(X_test)

    rmse = np.sqrt(mean_squared_error(y_test, y_pred)) 
    r2 = r2_score(y_test, y_pred)

    experiment.log_metrics({"rmse": rmse, "r2": r2}) 
    experiment.end()

    return {"rmse": rmse, "r2": r2}

Again we use CometML to log the model‘s performance metrics, in this case RMSE and R-squared. These will show up in the Comet user interface where we can easily compare different runs.

Step 6: Orchestrate

Finally, let‘s connect all our steps into an end-to-end pipeline. In the flow definition bulldozer_flow.py:

from prefect import Flow

from ingest import load_data
from clean import preprocess_data 
from split import split_data
from train import train_model
from evaluate import evaluate_model

DATA_PATH = "data/TrainAndValid.csv"
TARGET_COL = "SalePrice"

with Flow("bulldozer-flow") as flow:
    df = load_data(DATA_PATH)
    df_clean = preprocess_data(df)  
    X_train, X_test, y_train, y_test = split_data(df_clean, TARGET_COL)
    model = train_model(X_train, y_train)
    metrics = evaluate_model(model, X_test, y_test)

flow.run()  

We import our task functions and invoke them according to the pipeline logic. Prefect takes care of dependency management, so each task will only run once its upstream dependencies are complete. Calling flow.run() executes the pipeline and streams logs to the console.

With a few additional lines, we can also register our flow with Prefect Cloud to unlock scheduling, versioning, and other production features:

from prefect.deployments import DeploymentSpec
from prefect.orion.schemas.schedules import IntervalSchedule
from prefect.flow_runners import SubprocessFlowRunner
from datetime import timedelta

DeploymentSpec(
    name="bulldozer-model-training",
    flow=flow,
    schedule=IntervalSchedule(interval=timedelta(days=1)),
    flow_runner=SubprocessFlowRunner(),
    tags=["ml"]
)

This schedules our workflow to run daily and execute in a subprocess to isolate packages. A huge benefit of Prefect is that it abstracts away the orchestration details, allowing us to focus on the business logic.

Results and Insights

After running our pipeline, we can head over to the CometML panel to view the experiment results. Some key findings:

• The LightGBM model achieved an RMSE of 5324 and R^2 of 0.89 on the test set, indicating strong predictive performance.
• Feature importance analysis shows that YearMade, Machine_Hours_Current_Meter, and ProductSize were the most informative features for price prediction.
• A partial dependence plot revealed that bulldozers with ModelYear > 2010 fetch significantly higher prices.

CometML makes it easy to discover and share these insights from your experiment runs. Having this information easily accessible is crucial for debugging, reporting, and guiding future development.

Next Steps

In this tutorial, we covered the core steps for implementing an end-to-end MLOps pipeline with Prefect and CometML. Some possible extensions:

• Expand the modeling code to include hyperparameter tuning
• Add a model validation step that checks for data drift
• Include model explainability analysis using SHAP or LIME
• Deploy the trained model as a REST API endpoint
• Set up Prefect Slack notifications on failure

Both Prefect and CometML are evolving rapidly with exciting roadmaps:

• Prefect 2.0, currently in beta, introduces a fresh Orion UI and fully reworked Core library with async support
• CometML continues to expand its model monitoring capabilities, with new integrations for real-time inference logging and monitoring

Conclusion

Prefect and CometML form a powerful stack for building production-grade MLOps pipelines. With Prefect, you can easily orchestrate complex workflows, enabling better collaboration between data and ML engineering teams. And CometML provides critical experiment tracking and model monitoring capabilities to maintain performance and catch issues proactively.

While not a complete MLOps solution, these two tools significantly boost productivity vs a bespoke approach. I encourage you to try them out on your own projects and see the results for yourself!