ML engineering pipelines

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…

dagster ml-pipeline asset

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…

csv ml-pipelines io-stringio

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…

14 0 Open

How to Mock MLflow log_params and log_metrics in Python

Use unittest.mock to patch MLflow's log_param and log_metric, run the training function, and verify logging calls without touching a real tracking server.

mlflow mock testing

from unittest.mock import Mock, patch
import mlflow


def train_model():
    mlflow.log_param("learning_rate", 0.01)
    mlflow.log_param("epochs", 10)
    mlflow.log_metric("accuracy", 0.95)
    mlflow.log_metric("loss", 0.05)
    return "Training completed"


if __name__ == "__main__":
    with patch("mlflow.log_par…

15 0 Open

train_test_split mock unit-testing

How to Mock train_test_split in Python for Unit Testing

Build a lightweight mock of sklearn's train_test_split to unit test ML pipeline code without needing the full library or deterministic random state.

import numpy as np
from sklearn.model_selection import train_test_split
from unittest.mock import patch

def mock_train_test_split(X, y, test_size=0.25, random_state=None, **kwargs):
    """A simple mock implementation of train_test_split."""
    n_samples = len(X)
    n_test = int(n_samples * test_size)
    n_train =…

12 0 Open

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.

numpy batch ml

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…

14 0 Open

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.

airflow ml pipeline

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 …

ml drift-detection retraining

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)
       …

16 0 Open

csv data-loading standard-library

Load CSV Training Data Without Pandas in Python

This code loads a CSV file into a list of dictionaries using only the standard library, ideal for small ML training data without heavy dependencies.

import csv
from pathlib import Path

def load_csv(path):
    """Load CSV file into list of dicts without pandas."""
    rows = []
    with open(path, newline='', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        for row in reader:
            rows.append(dict(row))
    return rows

if __name__ == "__m…