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.
Python code
47 linesimport 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)
def process_batch(self):
batch = []
batch_end = time.time() + self.batch_interval
while time.time() < batch_end and self.source:
batch.append(self.source.popleft())
if batch:
self.processed.extend(batch)
print(f"[{datetime.now().strftime('%H:%M:%S')}] "
f"processed {len(batch)} events: {batch}")
else:
print(f"[{datetime.now().strftime('%H:%M:%S')}] empty batch")
return batch
if __name__ == "__main__":
stream = MicroBatchStream(batch_interval_sec=1)
# Simulate incoming events over time
stream.add_events(["click", "view", "purchase"])
time.sleep(0.3)
stream.add_events(["login", "logout"])
time.sleep(0.5)
stream.add_events(["search", "filter"])
# Run 3 micro-batches
for _ in range(3):
batch = stream.process_batch()
time.sleep(0.5)
print(f"\nTotal processed: {len(stream.processed)}")
print(f"Remaining in source: {len(stream.source)}")Output
[14:23:45] processed 3 events: ['click', 'view', 'purchase']
[14:23:46] processed 2 events: ['login', 'logout']
[14:23:47] processed 2 events: ['search', 'filter']
Total processed: 7
Remaining in source: 0How it works
The MicroBatchStream class mimics Spark Streaming's micro-batch model by accumulating events in a deque and draining them every batch_interval seconds. process_batch() blocks until the interval elapses or the source is empty, then moves available events to the processed list and prints a log line. This simulation is useful for testing stream-processing logic without a Spark cluster. The time-based loop uses time.time() to ensure batches are sized by wall-clock time, not CPU iterations.
Common mistakes
- Assuming `deque.popleft()` is thread-safe; it is not, so use a lock in multi-threaded scenarios.
- Using `time.sleep()` inside the batch loop increases latency; prefer a non-blocking check or a timer.
- Not handling the case where `source` is empty but the batch window hasn't elapsed, leading to wasted wait time.
- Forgetting that `deque` is not sorted; events are processed in insertion order, which may not match event timestamps.
Variations
- Use `asyncio` with `asyncio.sleep` for an asynchronous micro-batch stream.
- Replace `deque` with `queue.Queue` to support multiple producer threads.
Real-world use cases
- Unit-testing Spark Structured Streaming transformations locally without spinning up a Spark session.
- Load-testing downstream sinks with predetermined event bursts before deploying to production.
- Debugging checkpoint and stateful aggregation logic by replaying event logs through a mock stream.
Sponsored
More from Big data & Spark
Keep learning
Related tutorials and quizzes for this topic.