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 = …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 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 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 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 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 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…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.