Big data & Spark
PySpark jobs, partitioning, batch processing, and large-dataset transform patterns.
Accumulators Global Counter Mock in Python
Shows an accumulator-style global counter with a mock patch to control its value in tests.
import unittest
from unittest.mock import patch
# Module-level global counter accumulator
counter = 0
def increment(by=1):
"""Increment the global counter in place (accumulator pattern)."""
global counter
counter += by
return counter
def reset():
"""Reset the counter to zero."""
global count…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…Cache persist MEMORY_ONLY mock in Python
Mock a MEMORY_ONLY persistence cache in Python with an LRU eviction policy and optional persistence flag.
import time
class LRUCache:
def __init__(self, capacity, persistence="MEMORY_ONLY"):
self.capacity = capacity
self.persistence = persistence
self.cache = {}
self.access_order = []
self.hits = 0
self.misses = 0
def get(self, key):
if key in self.cache:
…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 Broadcast a Small Lookup Table in Python
Simulates broadcasting a small lookup table by iterating key-value pairs and emitting packed rows to subscribers with deterministic output.
import random
# Generate a deterministic mock broadcast of a small lookup table
# with 5 keys and random integer values (seeded for reproducibility)
data = {
"sensor_a": 22,
"sensor_b": 87,
"sensor_c": 43,
"sensor_d": 65,
"sensor_e": 31,
}
# Simulate a broadcast to subscribers by iterating and p…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 Explode an Array Column in Python
This code demonstrates a mock explode operation that converts an array column into multiple rows, similar to Spark's explode function.
import json
def explode_array_column(data, column):
"""Mock explode: split array column into multiple rows."""
exploded = []
for row in data:
values = row.get(column, [])
for value in values:
new_row = dict(row)
new_row[column] = value
exploded.append(n…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 Partition Pruning in Python
A dataclass-based mock that filters partitions by year and month to emulate Spark's partition pruning logic.
from dataclasses import dataclass
from typing import List
@dataclass(frozen=True)
class Partition:
id: int
year: int
month: int
class PartitionPruner:
"""Mock partition pruning: only keep partitions that match the filter."""
def __init__(self, partitions: List[Partition]):
self._partiti…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 Compute-Collect Action Trigger in Python
Mock a compute-collect action trigger using Python's unittest.mock to simulate Spark-style job execution and assert trigger behavior.
Here's a Python code sample for the problem title "Action trigger compute collect mock":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…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.