Reference library

ML engineering pipelines

Feature prep, batch inference, model-serving hooks, and production ML workflow glue.

4 matches

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.

data drift psi monitoring

Python

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…

13 0 Open

ML engineering pipelines medium

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.

kubeflow pipelines mlops

Python

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…

14 0 Open

ML engineering pipelines medium

How to Stage ML Model Workflows with Python Classes

Defines a Stage class to model ML pipeline stages with variants and mocks, printing grammar for Model, Staging, and Production stages.

ml-pipelines stages model-deployment

Python

class Stage:
    def __init__(self, name):
        self.name = name
        self.mocks = []
        self.variants = []

    def add_mock(self, mock_name):
        self.mocks.append(mock_name)

    def add_variant(self, variant_name, productions=()):
        self.variants.append((variant_name, list(productions)))

    …

12 0 Open

ML engineering pipelines medium

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.

ml-pipelines mock tempfile

Python

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…

13 0 Open

Browse by section

Each section groups closely related Python snippets.