Big data &amp; Spark

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…

16 0 Open

Big data & Spark easy

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.

pyspark hive mock

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

apache-spark logical-plan catalyst

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…

13 0 Open

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.

testing mock spark

Here's a Python code sample for the problem title "Action trigger compute collect mock":

Big data & Spark easy

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.

socket mock streaming

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…

Big data & Spark easy

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.

udf mock testing

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, *…

13 0 Open

rate-limiting mock-testing streaming

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…

16 0 Open

pyspark structured-streaming foreachbatch

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…