Python Code
Samples
Copy-ready Python snippets by topic and difficulty — short, focused, and runnable in the browser editor.
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.
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"…
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.
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…
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.
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:
…
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.
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…
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.
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)
…
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.
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…
Session window gap mock in Python
Group sorted timestamps into sessions where any gap between consecutive events exceeds a threshold starts a new session.
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…
Sliding Window Streaming Mock in Python
A simple Python class that maintains a sliding window of recent streaming values and computes the running average.
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…
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 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.
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…
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.
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)))
…
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 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 *…
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…
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…
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…
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.
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 +…
Bootstrap Confidence Interval in Python
Estimates a confidence interval for a statistic (like the mean) using bootstrap resampling in pure 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…
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.
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)
…
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.
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…
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.
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)
…
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.
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…
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
- Pick a topic section — strings, lists, files, functions, and more
- Open a sample, read How it works, and copy the code block
- 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.