Big data & Spark
PySpark jobs, partitioning, batch processing, and large-dataset transform patterns.
How to Mock a Hash Join on Large and Small Tables in Python
This code efficiently joins a large dataset (1000 rows) with a small lookup table (20 rows) by building a dictionary hash lookup, mimicking a hash join strategy used in big data systems.
import random
from pprint import pprint
# Large table: 1000 rows (id, group_id, value)
large = [{"id": i, "group_id": random.randint(1, 20), "value": random.random() * 100} for i in range(1000)]
# Small table: 20 rows (group_id, label)
small = [{"group_id": g, "label": f"Group-{g}"} for g in range(1, 21)]
# Mock a …How to Mock a Parquet partitionBy Sink in Python
Manually write a DataFrame to partitioned Parquet files, mimicking Spark's partitionBy sink behavior without Spark.
import pyarrow as pa
import pyarrow.parquet as pq
from pathlib import Path
import tempfile
import shutil
def mock_partition_by_sink(data, output_dir, partition_cols):
table = pa.Table.from_pandas(data)
schema = table.schema
unique_combos = table.select(partition_cols).to_pylist()
seen = set()
for…How to Mock a Socket Stream in Python
Simulate a streaming socket source with a generator to test stream-read and buffering logic without a real network.
import socket
import threading
import time
def mock_socket_stream(data_chunks, delay=0.1):
"""Generator that simulates a streaming socket source."""
for chunk in data_chunks:
time.sleep(delay)
yield chunk
def read_stream_socket(stream_gen):
"""Reads from mock stream and prints received ch…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 Pivot and Group Aggregate in Python
Group records by a key, collect values, and apply an aggregate function (like sum) to build a pivot-style summary dictionary.
from collections import defaultdict
def pivot_group_aggregate(records, group_key, value_key, agg_func):
groups = defaultdict(list)
for record in records:
groups[record[group_key]].append(record[value_key])
return {key: agg_func(values) for key, values in groups.items()}
if __name__ == "__main__":…How to Shuffle Items by Group in Python
Randomly shuffle items within each group while keeping groups contiguous, using a seed for reproducible results.
import random
def shuffle_sort_groups(items, group_key, seed=None):
"""Randomize order within groups, keeping groups contiguous."""
rng = random.Random(seed)
groups = {}
for item in items:
key = group_key(item)
groups.setdefault(key, []).append(item)
result = []
for k…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 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 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 select specific columns in Python with SQLite
A reusable function that connects to a SQLite database and returns only the requested columns from a given table.
import sqlite3
def select_pruned_columns(db_path, table, columns):
with sqlite3.connect(db_path) as conn:
cursor = conn.cursor()
col_list = ", ".join(columns)
query = f"SELECT {col_list} FROM {table}"
return cursor.execute(query).fetchall()
if __name__ == "__main__":
conn = sq…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)
…HyperLogLog Cardinality Estimation in Python
A small HyperLogLog implementation using MD5 hashing and 256 registers to estimate the number of unique items in a large stream with fixed memory.
import hashlib
import math
class HyperLogLog:
def __init__(self, b=8):
self.b = b
self.m = 1 << b
self.registers = [0] * self.m
self.alpha = 0.7213 / (1 + 1.079 / self.m)
def add(self, item):
h = int(hashlib.md5(str(item).encode()).hexdigest(), 16)
idx = h & (s…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…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…Skew Join Salting Key in Python (Demo)
Demonstrates skew join salting by expanding a smaller side with salt keys and matching rows on the larger side via random salt assignment.
import random
def skew_join_salting_key(left_df, right_df, salt_range=4):
"""
Demonstrates skew join salting: expand the smaller side with salt keys,
then attach a salt key to each row on the larger side.
Returns a list of (left, right, salt) tuples.
"""
skewed_left = []
for row in left_d…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…Browse by section
Each section groups closely related Python snippets.
Big data & Spark — Python code examples
What you will find here
This page collects big data & spark snippets — short, copy-ready Python you can paste into our free online IDE and run without installing anything. Each sample includes a plain-English explanation and the full source code.
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.