ML engineering pipelines
Feature prep, batch inference, model-serving hooks, and production ML workflow glue.
How to Mock a Feature Store Online Lookup in Python
This code simulates an online feature store with single and batch retrieval methods, using a dict-backed cache and timestamps.
import random
import time
class OnlineFeatureStore:
def __init__(self):
self.features = {}
def put(self, entity_id: str, feature_name: str, value):
key = (entity_id, feature_name)
self.features[key] = (value, time.time())
def get(self, entity_id: str, feature_name: str):
…How to Mock a Kubeflow Pipeline in Python
Build a minimal in-memory mock of a Kubeflow pipeline DAG using dataclasses and OrderedDict to chain component functions.
from typing import Dict, Any
from dataclasses import dataclass, field
from collections import OrderedDict
@dataclass
class KubeflowPipelineMock:
"""A minimal mock of a Kubeflow pipeline DAG."""
name: str
components: OrderedDict[str, callable] = field(default_factory=OrderedDict)
def add_component(se…How to Run Batch Predictions with a Mock Model in Python
Build a lightweight mock model class and run predictions across a batch of samples, returning results as a plain Python list.
import numpy as np
class MockModel:
def __init__(self, weights):
self.weights = np.array(weights)
def predict(self, X):
return X @ self.weights
def predict_batch(model, batch):
"""Run predictions for a batch of samples and return results as a list."""
return model.predict(np.array(ba…How to Save and Load a Mock Model with Pickle and joblib in Python
Serialize a custom machine learning model to a .joblib file with joblib.dump, reload it, and run a prediction with joblib.load.
import joblib
from pathlib import Path
class MockModel:
def __init__(self, weights):
self.weights = weights
def predict(self, features):
return sum(w * f for w, f in zip(self.weights, features))
def save_model_pickle(model, filepath):
with open(filepath, "wb") as f:
joblib.dump(…How to Simulate an Airflow ML Pipeline in Python
Mock an Airflow ML pipeline in plain Python by defining steps, simulating their execution with delays, and returning a success summary.
from datetime import datetime, timedelta
import time
class MLPipeline:
def __init__(self, pipeline_name):
self.pipeline_name = pipeline_name
self.steps = []
def add_step(self, step_name, duration_seconds):
self.steps.append({"name": step_name, "duration": duration_seconds})
def …How to Train a Gradient Boosting Regressor in Python
Build and evaluate a scikit-learn GradientBoostingRegressor on a synthetic dataset, printing test MSE and feature importances.
import numpy as np
from sklearn.ensemble import GradientBoostingRegressor
from sklearn.metrics import mean_squared_error
def train_gradient_boosting_mock():
# Toy regression dataset
np.random.seed(42)
X = np.random.rand(100, 3) * 10
y = 2 * X[:, 0] - 1.5 * X[:, 1] + 0.5 * X[:, 2] + np.random.normal(0,…How to Trigger Model Retraining on Drift in Python
Automatically detects accuracy drift in a mock ML model and triggers retraining when performance falls below a threshold.
import random
import time
class MockModel:
def __init__(self, name):
self.name = name
self.accuracy = 0.85
self.version = 1
def train(self, data_size):
# Simulate training time and accuracy improvement
time.sleep(0.1)
drift = random.uniform(-0.02, 0.02)
…How to do feature selection with VarianceThreshold in Python
This code demonstrates how to use scikit-learn's VarianceThreshold to remove low-variance features from a NumPy array, keeping only those that vary enough to be useful for modeling.
import numpy as np
from sklearn.feature_selection import VarianceThreshold
def main():
# Mock dataset: 4 samples, 5 features
X = np.array([
[0.1, 0.2, 1.0, 1.0, 0.5],
[0.2, 0.2, 0.0, 1.0, 0.4],
[0.1, 0.2, 1.0, 1.0, 0.6],
[0.3, 0.2, 1.0, 0.0, 0.5]
])
# Select features w…How to mock an artifact store with local paths in Python for ML pipelines
Create a temporary local artifact store with dummy files and metadata to test ML pipeline code without real storage.
import tempfile
from pathlib import Path
import json
def create_artifact_store_mock(base_path: Path = None):
"""Create a local artifact store mock directory structure."""
if base_path is None:
base_path = Path(tempfile.mkdtemp())
store_layout = {
"artifacts": [
{"name": "mode…How to ordinal encode categorical data in Python with sklearn
Convert job title categories into ordinal numeric labels using sklearn's OrdinalEncoder with explicit ordering.
from sklearn.preprocessing import OrdinalEncoder
import numpy as np
# Mock data: small job title categories with known ordering
data = np.array([
["intern"],
["junior"],
["mid"],
["senior"],
["lead"]
])
# Define the ordinal order (lowest to highest)
categories = [["intern", "junior", "mid", "seni…K-Fold Cross Validation in Python: A Simple Implementation
Implements k-fold cross validation from scratch, splitting data into folds and computing MSE scores for a baseline mean-predictor model.
import random
from statistics import mean
def cross_validation_scores(data, labels, k=5, seed=42):
random.seed(seed)
indices = list(range(len(data)))
random.shuffle(indices)
fold_size = len(indices) // k
folds = []
for i in range(k):
if i == k - 1:
folds.append(indices[i *…Model registry version mock in Python
A simple in-memory model registry that stores model versions with metadata and supports version listing and latest retrieval.
class ModelRegistry:
def __init__(self):
self.models = {}
def register(self, name, version, model_type, metrics=None):
if name not in self.models:
self.models[name] = []
entry = {
"version": version,
"model_type": model_type,
"metrics": m…One Hot Encode Categories in Python
Convert a list of categorical strings into one-hot encoded numeric vectors using pure Python and NumPy.
import numpy as np
categories = ["red", "green", "blue", "red", "blue", "green", "red"]
unique = sorted(set(categories))
lookup = {cat: i for i, cat in enumerate(unique)}
one_hot = []
for cat in categories:
row = [0] * len(unique)
row[lookup[cat]] = 1
one_hot.append(row)
print("Categories:", categories…StandardScaler mock in Python
A pure-Python StandarScaler class that standardizes features to zero mean and unit variance without sklearn.
import math
class StandardScaler:
def __init__(self):
self.mean_ = None
self.std_ = None
def fit(self, X):
n = len(X)
self.mean_ = [sum(col) / n for col in zip(*X)]
self.std_ = []
for col in zip(*X):
variance = sum((x - self.mean_[i]) ** 2 for i, x …Train Logistic Regression From Scratch in Python
Trains a binary logistic regression model using gradient descent on mock data, printing learned weights and probabilities.
import numpy as np
# Mock data: 2 features, binary classification
X = np.array([[1, 2], [2, 3], [3, 4], [4, 5], [5, 6]])
y = np.array([0, 0, 1, 1, 1])
# Add bias term (column of ones)
X_b = np.c_[np.ones((X.shape[0], 1)), X]
# Initialize parameters
theta = np.zeros(X_b.shape[1])
# Hyperparameters
learning_rate = 0…Browse by section
Each section groups closely related Python snippets.
ML engineering pipelines — Python code examples
What you will find here
This page collects ml engineering pipelines snippets — short, copy-ready Python you can paste into our free online IDE and run without installing anything. Each sample includes a plain-English explanation and the full source code.
Samples vs tutorials and challenges
Samples are quick reference — one concept per page. For step-by-step teaching, use our Python tutorials. To test yourself, try quizzes or coding challenges. Clean up style with the Python formatter.