Python Code
Samples
Copy-ready Python snippets by topic and difficulty — short, focused, and runnable in the browser editor.
How to Mock a UDAF Aggregate Function in Python
This code provides a minimal mock of a User-Defined Aggregate Function (UDAF), simulating the initialize-update-merge-finalize lifecycle with a defaultdict counter.
from collections import defaultdict
class MockUDAF:
"""A minimal mock of a User-Defined Aggregate Function.
Simulates aggregate lifecycle: initialize, update per row,
and finalize the result.
"""
def __init__(self):
self._buffer = defaultdict(int)
def initialize(self):
"""Re…How to Mock a User-Defined Function (UDF) in Python
Wrap a real UDF implementation with call logging to simulate and track invocations in a data pipeline.
from typing import Any, Callable
# Mock a user-defined function (UDF) that was previously complex or external
def mock_udf(name: str, implementation: Callable[..., Any], *, calls: list[Any]) -> Callable[..., Any]:
"""Wrap a real implementation with call logging to simulate a UDF."""
def wrapper(*args: Any, *…How to Mock and Test a Rate-Limited Source Stream in Python
Build a class that rate-limits emitted items using a sliding window and test it with a simulated stream in Python.
import time
from collections import deque
class RateLimitedSource:
def __init__(self, max_rate, window=1.0):
self.max_rate = max_rate
self.window = window
self._timestamps = deque()
def emit(self, item):
now = time.monotonic()
while self._timestamps and self._timestam…How to Simulate a MapReduce Mock with Combine Phase in Python
Simulates a MapReduce pipeline with a combiner that aggregates local counts per reducer to reduce network and compute overhead.
from collections import defaultdict
def map_phase(lines):
intermediate = defaultdict(list)
for line in lines:
for word in line.strip().lower().split():
intermediate[word].append(1)
return dict(intermediate)
def combine_phase(intermediate, num_reducers=3):
combined = defaultdict(li…How to Use Broadcast Variables as Read-Only in PySpark (Mock Example)
Share a lookup dict across Spark executors with a broadcast variable and verify its read-only behavior in a local mock.
from pyspark import SparkContext, SparkConf
def main():
conf = SparkConf().setAppName("BroadcastMock").setMaster("local[2]")
sc = SparkContext(conf=conf)
lookup = {"a": 1, "b": 2, "c": 3}
broadcast_lookup = sc.broadcast(lookup)
data = ["a", "b", "c", "a", "unknown"]
rdd = sc.parallel…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)
…Lazy Evaluation Transform Lineage Mock in Python
Build a mock lineage tracker for data transforms using lazy evaluation and function wrappers in Python.
import functools
def lazy_transform(pipeline):
"""Build a mock lineage tracker using lazy evaluation."""
lineage = []
def wrap(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
result = func(*args, **kwargs)
lineage.append({"transform": func.__name__, "a…Mock Predicate Pushdown in Python for Big Data Queries
Simulate predicate pushdown by applying filters at the storage layer before materializing rows, showing how big data engines optimize queries.
class Query:
def __init__(self, table, rows):
self.table = table
self.rows = rows
def filter(self, predicate):
return Query(
self.table,
[row for row in self.rows if all(predicate(row) for predicate in predicate)]
)
def filter_pushdown(self, predica…Mock RDD in Python: Simulate Spark RDD Lazy Transformations
Simulate Apache Spark RDD behavior in Python with lazy maps, filters, partitions, and a collect action.
import random
def mock_rdd(data, num_slices=2):
"""
A simple simulation of Spark RDD behavior with lazy evaluation,
transformations, and an action.
"""
class SimpleRDD:
def __init__(self, data, num_slices=2):
self.data = data
self.num_slices = num_slices
…Modeling a Hive Metastore Table Schema in Python
A dataclass that mimics a Hive metastore table schema—columns, partition keys, storage format, and location—with helper methods for description and mutation.
from dataclasses import dataclass, field
from typing import Dict, List, Optional
@dataclass
class HiveTable:
"""Simple mock of a Hive metastore table schema."""
name: str
database: str = "default"
columns: List[Dict[str, str]] = field(default_factory=list)
partition_keys: List[Dict[str, str]] = f…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…Z-Order Optimization in Python
A mock concept demonstrating z-order layout optimization by reassigning z-indices based on areas size.
class ZOrderLayout:
"""
Minimal mock for z-order layout optimization using a stacking score.
Elements overlap; higher z_index is drawn on top.
"""
def __init__(self):
self.elements = []
def add_element(self, name, area, z_index):
self.elements.append({"name": name, "area": area…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 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):
…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…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.