Big data & Spark
PySpark jobs, partitioning, batch processing, and large-dataset transform patterns.
Approximate Distinct Count in Python with HyperLogLog
Mock a large data stream and estimate the number of distinct items with a HyperLogLog-style probabilistic counter to save memory.
import random
import string
from collections import Counter
import math
class ApproxCountDistinct:
def __init__(self, num_buckets=16):
self.num_buckets = num_buckets
self.max_zeros = [0] * num_buckets
def _hash(self, item):
# Simple string hash to a 32-bit integer
h = …Bloom Filter Join Mock in Python
A mock hash join that uses a Bloom filter to pre-filter one table before performing an exact match, reducing the number of comparisons in large dataset joins.
import hashlib
import random
import string
class BloomFilter:
def __init__(self, size: int = 200, num_hashes: int = 3):
self.bits = [False] * size
self.size = size
self.num_hashes = num_hashes
def _hashes(self, item: str):
result = []
for seed in range(self.num_hashes…Compaction Small Files Mock in Python
Simulates a small-files compaction job by creating small mock files and merging them into a single output file using Python's standard library.
from pathlib import Path
import tempfile
import os
def create_small_files(directory: Path, file_count: int = 5, lines_per_file: int = 3):
"""Create several small mock files with sample content."""
directory.mkdir(exist_ok=True)
for i in range(file_count):
file_path = directory / f"part-{i:04d}.tx…Delta Lake ACID Transaction Log Mock in Python
Simulates Delta Lake's transactional log with JSON files for atomic commits, versioned operations, and crash recovery
import json
import time
from pathlib import Path
class DeltaLog:
def __init__(self, path):
self.log_dir = Path(path)
self.log_dir.mkdir(parents=True, exist_ok=True)
self.version = 0
def _write_txn(self, action, payload):
txn = {
"version": self.version,
…How to Build a DAG Execution Stage Calculator in Python
Computes the execution stages of a directed acyclic graph (DAG) by grouping nodes that become ready simultaneously using topological sorting with Kahn's algorithm.
from collections import defaultdict, deque
def get_stages(edges):
"""Return list of stages, where each stage is a list of nodes
that become ready at the same time in a DAG."""
graph = defaultdict(list)
in_degree = defaultdict(int)
nodes = set()
for src, dst in edges:
graph[src].appen…How to Create a Mock Iceberg Snapshot Manifest in Python
Build a mock Iceberg snapshot manifest structure with metadata and data entries using Python dictionaries and JSON.
import json
from datetime import datetime, timezone
def create_mock_manifest(snapshot_id: int, file_paths: list[str]) -> dict:
"""Create a mock Iceberg snapshot manifest structure."""
manifest_file = {
"manifest_path": f"/warehouse/table/metadata/snap-{snapshot_id}-m0.avro",
"manifest_length"…How to Create a Mock Kafka Producer in Python
Build a Kafka producer that generates mock streaming records with JSON serialization and error handling for local testing.
import json
import time
from kafka import KafkaProducer
from kafka.errors import KafkaError
def create_mock_producer(bootstrap_servers="localhost:9092", topic="input-topic"):
"""Create a Kafka producer that generates mock streaming data."""
producer = KafkaProducer(
bootstrap_servers=bootstrap_servers…How to Filter and Project Spark DataFrames with PySpark SQL
Simulate a SQL SELECT with WHERE using PySpark DataFrame select and filter to project columns and apply conditions.
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
spark = SparkSession.builder.appName("QueryFilterMock").master("local[2]").getOrCreate()
data = [
("Alice", 28, "Engineering"),
("Bob", 35, "Sales"),
("Carol", 32, "Engineering"),
("David", 25, "Marketing"),
("Eve", 29, "E…How to Implement MapReduce Word Count in Python Using a Dict
Simulate a MapReduce word count pipeline in Python with a mock dict, splitting text into words, shuffling, and reducing to frequency counts.
def map_reduce_word_count(text: str) -> dict:
"""Simulate a MapReduce pipeline to count word frequencies."""
# MAP phase: split into words and emit (word, 1) pairs
mapped = []
for word in text.lower().split():
# Clean word of punctuation
clean_word = ''.join(char for char in word if cha…How to Implement a Mock MapReduce for Word Count in Python
Simulates a MapReduce word count pipeline with mapper, shuffle, and reducer phases using Python dicts and standard library modules.
from collections import defaultdict
import re
def mapper(text):
"""Split text into words and emit (word, 1) pairs."""
words = re.findall(r'\b\w+\b', text.lower())
return [(word, 1) for word in words]
def reducer(pairs):
"""Group word-count pairs and sum counts."""
counts = defaultdict(int)
fo…How to Implement a Streaming Watermark in Python
Mock structured streaming watermarks in Python to track late event times and compute a watermark for windowed processing.
from datetime import datetime, timedelta
import time
class StreamingWatermark:
"""Mock watermark tracker for structured streaming."""
def __init__(self, watermark_delay_seconds):
self.watermark_delay = timedelta(seconds=watermark_delay_seconds)
self.max_event_time = None
def observe_even…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 Implement row_number Window Function in Python
This code implements a SQL-style ROW_NUMBER() window function in pure Python, partitioning rows by a set of columns and ranking them within each partition by an ordered set of columns.
from collections import defaultdict
import itertools
def row_number(rows, partition_by, order_by):
partitions = defaultdict(list)
for index, row in enumerate(rows):
key = tuple(row[col] for col in partition_by)
partitions[key].append((index, row))
result = []
for key in partitions:
…How to Mock DataFrame Schema Columns in Python
Create an empty pandas DataFrame with only the specified column names to mock a schema before any data is loaded.
import pandas as pd
def mock_schema(columns):
return pd.DataFrame(columns=columns)
if __name__ == "__main__":
cols = ["name", "age", "city"]
df = mock_schema(cols)
print(df)
print(f"Columns: {list(df.columns)}, Shape: {df.shape}")How to Mock Hive Support in PySpark with unittest.mock
This code demonstrates how to mock Hive support in a PySpark environment using unittest.mock to simulate SQL queries returning fixed data.
from unittest.mock import Mock, patch
def get_hive_tables(spark):
"""Mock Hive support by returning a fixed list of tables."""
return spark.sql("SHOW TABLES").collect()
class HiveTable:
"""Simple class that mimics a Hive table row."""
def __init__(self, database, tableName):
self.database =…How to Mock Spark Streaming Micro-Batches in Python
Simulate Spark's micro-batch streaming with a simple deque-based class that collects events over time and processes them in timed batches.
import time
from collections import deque
from datetime import datetime
class MicroBatchStream:
def __init__(self, batch_interval_sec=2):
self.batch_interval = batch_interval_sec
self.source = deque()
self.processed = []
def add_events(self, events):
self.source.extend(events…How to Mock a Catalyst Logical Plan in Python
Build a small Python class that mimics Spark Catalyst's logical plan tree for teaching or testing query optimizations.
from typing import Any, Dict, List, Optional
class CatalystLogicalPlan:
"""A minimal mock of Catalyst's logical plan for teaching purposes."""
def __init__(self, node_type: str, **kwargs: Any) -> None:
self.node_type = node_type
self.attributes: Dict[str, Any] = kwargs
self.child…How to Mock a File Source Watch Directory in Python
Poll a directory for new files and log changes, simulating a watch directory for data ingestion patterns.
import os
import time
from pathlib import Path
def watch_directory(dir_path: str, poll_interval: float = 1.0, max_iterations: int = 5):
"""
Mock a file-source watch directory by polling for changes.
Returns new files detected during each poll cycle.
"""
directory = Path(dir_path)
directory.mk…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…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.