Reference library

Python Code Samples

Copy-ready Python snippets by topic and difficulty — short, focused, and runnable in the browser editor.

506 matches
Big data & Spark easy

How to Implement collect_list in Python

Group rows by a key and collect all corresponding values into a list — a pure-Python mock of Spark's collect_list aggregation.

collect_list aggregation grouping
Python
from collections import defaultdict

def collect_list(rows, key_field, value_field):
    grouped = defaultdict(list)
    for row in rows:
        grouped[row[key_field]].append(row[value_field])
    return dict(grouped)

if __name__ == "__main__":
    data = [
        {"dept": "sales", "emp": "alice"},
        {"dept"…
18 0 Open
Big data & Spark easy

How to Truncate Lineage Back to a Checkpoint in Python

Walks a linked list of lineage nodes upward to find the nearest checkpoint and returns that node, truncating the lineage.

lineage checkpoint linked-list
Python
class LineageNode:
    def __init__(self, name, parent=None, checkpoint=None):
        self.name = name
        self.parent = parent
        self.checkpoint = checkpoint

    def truncate_at_checkpoint(self):
        """Truncate lineage back to the last checkpoint."""
        current = self
        while current.check…
18 0 Open
Big data & Spark medium

How to implement a tumbling window aggregation in Python

Build a mock tumbling window aggregator in Python that groups streaming events into fixed time intervals and computes count, sum, and average per window.

tumbling-window streaming aggregation
Python
import time
from collections import deque

class TumblingWindow:
    def __init__(self, duration_seconds):
        self.duration = duration_seconds
        self.buffer = deque()
        self.window_start = None

    def add(self, item):
        current_time = time.time()
        if self.window_start is None:
         …
15 0 Open
Big data & Spark medium

How to use foreachBatch with a mock sink in PySpark

Demonstrates using Spark Structured Streaming's foreachBatch sink to capture and verify streaming batches by writing them into a custom mock sink object.

pyspark structured-streaming foreachbatch
Python
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, lit

class MockSink:
    def __init__(self):
        self.batches = []
    
    def write_batch(self, batch_df, batch_id):
        # Collect batch data as list of dicts for verification
        records = batch_df.collect()
        self.batches…
15 0 Open
Big data & Spark easy

Hudi Upsert Mock Copy on Write in Python

Simulates Apache Hudi's Copy-on-Write upsert behavior by merging update records into a deep copy of base records, replacing matches or appending new ones.

hudi upsert copy-on-write
Python
import copy
from typing import Dict, List, Any

def upsert_copy_on_write(base_records: List[Dict[str, Any]], updates: List[Dict[str, Any]], key_field: str = "id") -> List[Dict[str, Any]]:
    """Simulate Hudi Copy-on-Write upsert: merge updates into a copy of base records."""
    result = copy.deepcopy(base_records)
 …
17 0 Open
Big data & Spark easy

Partition Data by Hash Key Mod N in Python

Returns a partition index for a string key by hashing it with MD5 and taking modulo N, then groups sample keys into partitions.

hashing partitioning hashlib
Python
import hashlib


def partition_key(key: str, num_partitions: int) -> int:
    """Return partition index for key using MD5 hash mod N."""
    digest = hashlib.md5(key.encode()).hexdigest()
    return int(digest, 16) % num_partitions


if __name__ == "__main__":
    keys = ["alice", "bob", "carol", "dave", "eve"]
    nu…
14 0 Open
Big data & Spark easy

Session window gap mock in Python

Group sorted timestamps into sessions where any gap between consecutive events exceeds a threshold starts a new session.

timestamps sessions windowing
Python
from datetime import datetime, timedelta


def session_windows(timestamps, gap_seconds=300):
    """Group timestamps into sessions where gaps > gap_seconds start new sessions."""
    if not timestamps:
        return []

    # Sort timestamps chronologically to ensure correct windowing
    timestamps = sorted(timestam…
16 0 Open
Big data & Spark easy

Sliding Window Streaming Mock in Python

A simple Python class that maintains a sliding window of recent streaming values and computes the running average.

streaming sliding-window averages
Python
import time
import random

class StreamingMock:
    """Produces a stream of numbers using a sliding window."""
    
    def __init__(self, window_size=5):
        self.window = []
        self.window_size = window_size
        
    def push(self, value):
        """Add a value, sliding the window forward."""
        s…
13 0 Open
ML engineering pipelines easy

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.

zenml ml pipeline
Python
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…
13 0 Open
ML engineering pipelines medium

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.

sklearn pipeline columntransformer
Python
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…
17 0 Open
ML engineering pipelines easy

How to Save and Load PyTorch Model State Dict in Python

This code demonstrates how to save a PyTorch model's state dict to a file and load it back into a new model instance, verifying weights match.

pytorch state-dict model
Python
import torch
import torch.nn as nn

class SimpleNet(nn.Module):
    def __init__(self):
        super().__init__()
        self.fc1 = nn.Linear(4, 8)
        self.fc2 = nn.Linear(8, 2)

    def forward(self, x):
        x = torch.relu(self.fc1(x))
        return self.fc2(x)

if __name__ == "__main__":
    model = Simp…
15 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)))

    …
14 0 Open
ML engineering pipelines medium

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.

sklearn gradient-boosting regression
Python
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,…
14 0 Open
ML engineering pipelines easy

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.

ordinal-encoding sklearn categorical-data
Python
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…
19 0 Open
ML engineering pipelines medium

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.

cross-validation ml model-evaluation
Python
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 *…
18 0 Open
ML engineering pipelines easy

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.

csv data-loading standard-library
Python
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…
16 0 Open
ML engineering pipelines easy

One Hot Encode Categories in Python

Convert a list of categorical strings into one-hot encoded numeric vectors using pure Python and NumPy.

one-hot encoding categorical numpy
Python
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…
16 0 Open
ML engineering pipelines medium

Train Logistic Regression From Scratch in Python

Trains a binary logistic regression model using gradient descent on mock data, printing learned weights and probabilities.

logistic-regression machine-learning gradient-descent
Python
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…
15 0 Open
A/B testing & experimentation medium

Bayesian A/B Test Credible Interval in Python

Simulates A/B test data and computes posterior credible intervals and the probability that variant B outperforms A using Bayesian Beta-Binomial inference.

bayesian ab-testing credible-interval
Python
import numpy as np
from scipy import stats

# Simulated A/B test data
n_A = 1000
n_B = 1000
conversions_A = 120
conversions_B = 140

# Prior: Beta(1, 1) uniform
alpha_prior, beta_prior = 1, 1

# Posterior parameters
alpha_A = alpha_prior + conversions_A
beta_A = beta_prior + n_A - conversions_A
alpha_B = alpha_prior +…
18 0 Open
A/B testing & experimentation medium

Bootstrap Confidence Interval in Python

Estimates a confidence interval for a statistic (like the mean) using bootstrap resampling in pure Python.

bootstrap confidence-interval statistics
Python
import random


def bootstrap_ci(data, statistic, n_bootstraps=1000, ci_level=0.95, seed=42):
    random.seed(seed)
    n = len(data)
    boot_stats = []

    for _ in range(n_bootstraps):
        sample = [random.choice(data) for _ in range(n)]
        boot_stats.append(statistic(sample))

    boot_stats.sort()
    l…
18 0 Open
A/B testing & experimentation medium

Delta Method for Ratio Metrics in A/B Testing with Python

Computes the confidence interval for the difference between two ratio metrics using the delta method, with mock A/B test data.

delta-method ab-testing ratio-metrics
Python
import numpy as np
from scipy.stats import norm


def delta_method_ratio_delta(control: np.ndarray, treatment: np.ndarray, confidence: float = 0.95):
    """Estimate confidence interval for ratio metric using delta method.

    Args:
        control: numerator/denominator pairs from control group (n x 2 array)
       …
18 0 Open
A/B testing & experimentation medium

How to Create an Interrupted Time Series Mock in Python

Generate simulated interrupted time series data with a pre/post-intervention trend, level shift, and noise to test segmented regression models.

interrupted-time-series simulation numpy
Python
import numpy as np

# Mock interrupted time series data
np.random.seed(42)
n_pre = 50
n_post = 50
time = np.arange(0, n_pre + n_post)

# Pre-intervention: linear trend + noise
pre_trend = 0.05 * time[:n_pre] + np.random.normal(0, 0.5, n_pre)

# Post-intervention: new slope + level shift + noise
post_trend = 0.05 * tim…
16 0 Open
A/B testing & experimentation easy

How to Mock a Confidence Interval for a Proportion in Python

Simulate a Bernoulli sample and compute a 95% confidence interval for a proportion using the normal approximation in Python.

confidence-interval simulation statistics
Python
import random
import math

def mock_ci(n=100, p_true=0.5, z=1.96, seed=42):
    """Simulate a sample proportion and compute its 95% confidence interval."""
    random.seed(seed)
    successes = sum(1 for _ in range(n) if random.random() < p_true)
    p_hat = successes / n
    se = math.sqrt(p_hat * (1 - p_hat) / n)
  …
18 0 Open
A/B testing & experimentation medium

How to Perform Intent-to-Treat Analysis in Python

Runs an intent-to-treat analysis on mock A/B test data, comparing outcomes by initial group assignment with a t-test for significance.

ab-testing intent-to-treat statistics
Python
import pandas as pd
import numpy as np


def intent_to_treat_analysis(data):
    """Perform intent-to-treat (ITT) analysis.

    ITT compares outcomes based on initial treatment assignment,
    regardless of whether participants actually received the treatment.
    """
    # Create a copy to avoid mutating the origina…
14 0 Open

Browse by section

Each section groups closely related Python snippets.

Guide: free Python code samples library

Copy-ready Python snippets for learners and developers

PythonSkillset code samples are short, focused examples organised by topic and difficulty. Every snippet is server-rendered HTML — readable by search engines and easy to copy. Open any sample, read the notes, copy the code, then press Try in editor to run it in the browser with Pyodide.

How to use this library

  1. Pick a topic section — strings, lists, files, functions, and more
  2. Open a sample, read How it works, and copy the code block
  3. Run it in the IDE, tweak values, then take a related quiz or tutorial lesson

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.