ML engineering pipelines
Feature prep, batch inference, model-serving hooks, and production ML workflow glue.
Bayesian Optimization in Python: A Simplified Mock Implementation
A toy Bayesian optimization loop with a Gaussian process prior, expected improvement acquisition, and noisy sampling to find a function's minimum.
import random
import math
class BayesianOptimizer:
def __init__(self, noise=0.1):
self.noise = noise
self.observations = []
def objective(self, x):
return (math.sin(3*x) + 0.5*x) / (1 + x**2)
def gaussian_process_prior(self, x1, x2, length_scale=0.5):
return math.…Build a Data Helper Class in Python for ML Pipelines
A beginner-friendly Python class that summarizes, filters, and exports ML dataset rows as JSON.
from typing import List, Dict, Any
import json
class DataHelper:
"""Beginner-friendly helpers for ML data pipelines."""
def __init__(self, data: List[Dict[str, Any]]):
self.data = data
self.keys = list(data[0].keys()) if data else []
def summary(self) -> Dict[str, Any]:
"…Build a Mock Random Forest Classifier in Python
Create a simple random-forest-like classifier with random majority voting between trees, including fit, predict, and predict_proba methods.
import random
class MockRandomForest:
def __init__(self, n_trees=10, random_state=42):
self.n_trees = n_trees
self.random_state = random_state
self.classes_ = None
self._class_counts = None
random.seed(random_state)
def fit(self, X, y):
self.classes_ = sorted(…Champion Challenger Deployment Mock in Python
Simulates an A/B champion-challenger ML deployment workflow — comparing two mock model accuracies and deciding which to promote to production.
import random
import time
class ModelMocker:
def __init__(self, name="Model", accuracy=0.85):
self.name = name
self.accuracy = accuracy
def predict(self, data):
"""Simulate prediction with some randomness."""
time.sleep(0.005) # simulate compute time
return 1 if rando…Compare Model A vs Model B Metrics in Python
A script that simulates and compares metrics between two ML models, showing a formatted diff table for quick insight.
import random
def compare_a_b(samples=5):
"""Mock comparison of model A vs model B predictions."""
metrics = ["accuracy", "precision", "recall", "f1"]
print(f"{'Metric':<12}{'Model A':>10}{'Model B':>10}{'Diff':>10}")
print("-" * 42)
random.seed(42)
for metric in metrics:
a = round(r…Create a Minimal Great Expectations Suite Mock in Python
Build a small Python class that mimics a Great Expectations suite, storing and serializing column expectations as JSON.
import json
class GreatExpectationsSuite:
"""A minimal mock of a Great Expectations suite."""
def __init__(self, suite_name, expectations=None):
self.suite_name = suite_name
self.expectations = expectations or []
def add_expectation(self, expectation_type, column=None, kwargs=None):
…Detect Concept Drift in Python with a Simple Statistical Test
Detect concept drift by comparing the mean of recent data against a reference distribution using a z-score-like threshold.
import random
import statistics
def detect_drift(recent, reference, threshold=1.5):
ref_mean = statistics.mean(reference)
ref_std = statistics.stdev(reference)
recent_mean = statistics.mean(recent)
drift_score = abs(recent_mean - ref_mean) / (ref_std if ref_std > 0 else 1)
drifted = drif…Grid Search Hyperparameters in Python
Perform exhaustive grid search over hyperparameter combinations using itertools.product and a scoring function.
import itertools
def grid_search(param_grid, score_fn):
"""Perform exhaustive grid search over hyperparameter combinations."""
keys = param_grid.keys()
names = list(keys)
values = [param_grid[name] for name in names]
results = []
for combination in itertools.product(*values):
params =…How to Build a Data Validation Schema in Python
Create a lightweight validation schema using dataclasses and lambda validators to check fields in a dictionary.
import re
from dataclasses import dataclass, field
from typing import Any, Callable
@dataclass
class Field:
name: str
validator: Callable[[Any], bool]
required: bool = True
def validate(self, value: Any) -> bool:
if not self.required and value is None:
return True
return …How to Build a Mock ML Pipeline with Prefect in Python
Create a lightweight Prefect flow with mock preprocessing, training, and evaluation tasks to prototype an ML pipeline end-to-end.
from prefect import task, flow
from datetime import datetime
@task
def preprocess_data(raw_value: float) -> float:
"""Mock preprocessing: normalize the input value."""
return raw_value / 100.0
@task
def train_model(features: float) -> dict:
"""Mock training: return a fake model artifact."""
return …How to Build a Mock Offline Feature Store in Python
Build an in-memory mock of an offline feature store with a dict-based FeatureStore class for storing and retrieving ML features by entity ID.
from datetime import datetime
from collections import defaultdict
class FeatureStore:
"""Simple in-memory mock of an offline feature store."""
def __init__(self):
self._features = defaultdict(dict)
def ingest(self, entity_id, feature_name, value, timestamp=None):
ts = timestamp or datet…How to Build a Mock TFX Pipeline in Python
Simulate a TFX-style ML pipeline with simple Python functions to understand component orchestration, data flow, and artifact passing.
# Mock TFX pipeline to illustrate component orchestration
def CsvExampleGen(data_path):
"""Mock component: Simulates reading CSV data."""
print(f"ExampleGen: Reading from {data_path}")
return {"records": 100, "name": "examples"}
def StatisticsGen(example_artifact):
"""Mock component: Simulates genera…How to Build a Simple ML Pipeline with ZenML in Python
Build a mock machine learning pipeline with ZenML steps for data loading, training, and evaluation, and run it to print the final accuracy.
from zenml import pipeline, step
@step
def load_data() -> dict:
"""Simulate loading data from a source."""
return {"accuracy": 0.0, "loss": 1.0}
@step
def train_model(data: dict) -> dict:
"""Simulate training a model."""
data["accuracy"] = 0.95
data["loss"] = 0.1
return data
@step
def eva…How to Build an sklearn Pipeline with ColumnTransformer in Python
A mock example showing how to chain preprocessing and a regression model into a single sklearn Pipeline, scaling numeric features and one-hot encoding categorical features with ColumnTransformer.
import numpy as np
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.pipeline import Pipeline
from sklearn.linear_model import LinearRegression
# Mock dataset
X = np.array([[1, 'red'], [2, 'blue'], [3, 'red'], [4, 'green'], [5, 'blue']], dtype=o…How to Compute a Confusion Matrix in Python
Compute a multi-class confusion matrix from true and predicted labels using pure Python dictionaries and nested lists, then format it for readable output.
from collections import defaultdict
def compute_confusion_matrix(y_true, y_pred, labels):
"""Compute confusion matrix using Python dicts and nested lists."""
label_index = {label: i for i, label in enumerate(labels)}
matrix = [[0] * len(labels) for _ in range(len(labels))]
for true, pred in zip(y…How to Create a Mock Metaflow Flow in Python
Build a minimal Metaflow flow with two sequential steps that pass data between them using instance attributes.
from metaflow import FlowSpec, step, current
class MockFlow(FlowSpec):
"""A minimal Metaflow flow to demonstrate basic steps and branching."""
@step
def start(self):
self.category = "mock"
print(f"Start step for {self.category} flow")
self.next(self.process)
@step
def pr…How to Create a Mock ONNX Model in Python
Build and export a minimal mock ONNX model with a Reshape and Gemm layer using the onnx helper API.
import onnx
import numpy as np
from onnx import helper, TensorProto
def create_mock_model():
# Define input and output tensors
input_tensor = helper.make_tensor_value_info('input', TensorProto.FLOAT, [1, 3, 224, 224])
output_tensor = helper.make_tensor_value_info('output', TensorProto.FLOAT, [1, 10])
…How to Define Dagster ML Assets in Python
Define a chain of Dagster software-defined assets that compute raw features, normalized features, and predictions for an ML pipeline.
from dagster import asset
@asset
def raw_features():
return {"sepal_length": [5.1, 4.9, 6.2], "sepal_width": [3.5, 3.0, 3.4]}
@asset
def normalized_features(raw_features):
values = raw_features["sepal_length"]
mean = sum(values) / len(values)
std = (sum((x - mean) ** 2 for x in values) / len(values…How to Detect Data Drift with PSI in Python
Calculate the Population Stability Index (PSI) in Python to compare expected vs actual distributions and detect data drift in machine learning pipelines.
import numpy as np
def calculate_psi(expected, actual, buckets=10):
"""Calculate Population Stability Index (PSI) between two distributions."""
# Create bucket edges based on expected distribution percentiles
edges = np.percentile(expected, np.linspace(0, 100, buckets + 1))
edges[-1] = np.inf # Ensur…How to Do Random Search for Hyperparameter Tuning in Python
A mock random search that samples hyperparameter combinations from a grid and ranks them by a dummy score, with a reproducible seed.
import random
# Mock random search over a small hyperparameter grid
param_grid = {
"learning_rate": [0.001, 0.01, 0.1],
"batch_size": [16, 32, 64],
"num_layers": [1, 2, 3]
}
def random_search(grid, n_iter=5, seed=42):
"""Perform random search over a hyperparameter grid."""
random.seed(seed)
k…How to Evaluate Accuracy, Precision, and Recall in Python
Compute accuracy, precision, and recall for a binary classification model using scikit-learn's metrics functions.
from sklearn.metrics import accuracy_score, precision_score, recall_score
if __name__ == "__main__":
y_true = [0, 1, 1, 0, 1, 0, 1, 1]
y_pred = [0, 1, 0, 0, 1, 0, 1, 1]
accuracy = accuracy_score(y_true, y_pred)
precision = precision_score(y_true, y_pred)
recall = recall_score(y_true, y_pred)
…How to Generate Experiment Tracking Run IDs in Python
Generate unique experiment run IDs with timestamps and random suffixes for tracking ML pipeline executions.
import random
import string
import time
def generate_run_id(prefix="exp"):
timestamp = time.strftime("%Y%m%d_%H%M%S")
suffix = "".join(random.choices(string.ascii_lowercase + string.digits, k=6))
return f"{prefix}_{timestamp}_{suffix}"
if __name__ == "__main__":
# Simulate tracking three experiment r…How to Impute Missing Values with Mean in Python
Replace None values in a list with the mean of the existing values using Python's statistics module.
import statistics
from statistics import mean
def impute_mean(values):
"""Replace None with the mean of the non-None values."""
# Filter out None to compute the mean of existing values
valid = [v for v in values if v is not None]
if not valid:
return values # nothing to impute if all are Non…How to Load CSV Training Data in Python Without Pandas
Load CSV training data using Python's standard library and mock it with io.StringIO for testing, returning headers and rows as dictionaries.
import csv
from pathlib import Path
def load_csv_training_data(file_path: str | Path) -> tuple[list[str], list[dict[str, str]]]:
"""Load CSV training data and return headers plus rows as dictionaries."""
with open(file_path, mode="r", newline="", encoding="utf-8") as csv_file:
reader = csv.DictReader…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.