Stream Processing Deep Dive — Event Time, Watermarks & Exactly-Once

from datetime import datetime, timedelta
import random

def processing_time_window(events, window_size_seconds=10):
    """Simple processing-time windowing (naive)."""
    current_window = []
    window_start = datetime.now()

    for event in events:
        now = datetime.now()
        if (now - window_start).total_seconds() >= window_size_seconds:
            result = sum(e["value"] for e in current_window)
            print(f"[Processing] Window {window_start.time()}: sum={result}")
            current_window = []
            window_start = now
        current_window.append(event)

    # Flush remaining
    if current_window:
        result = sum(e["value"] for e in current_window)
        print(f"[Processing] Window {window_start.time()}: sum={result}")

# Simulate events with simulated arrival times
events = [{"event_time": None, "value": random.randint(1, 10)}
          for _ in range(15)]
processing_time_window(events, window_size_seconds=2)

from datetime import datetime, timedelta
from collections import defaultdict

def event_time_window(events, window_size_seconds=10):
    """Event-time based windowing (correct for late data)."""
    windows = defaultdict(list)

    for event in events:
        et = event["event_time"]
        # Assign to correct window based on event time
        window_key = et.timestamp() // window_size_seconds
        windows[int(window_key)].append(event)

    for key in sorted(windows.keys()):
        window_events = windows[key]
        result = sum(e["value"] for e in window_events)
        window_start = datetime.fromtimestamp(key * window_size_seconds)
        print(f"[EventTime] Window {window_start.time()}: sum={result}, "
              f"count={len(window_events)}")

# Events with actual timestamps, arriving out of order
base = datetime.now() - timedelta(seconds=30)
events = [
    {"event_time": base + timedelta(seconds=3), "value": 5},
    {"event_time": base + timedelta(seconds=7), "value": 3},
    {"event_time": base + timedelta(seconds=1), "value": 8},  # Late!
    {"event_time": base + timedelta(seconds=12), "value": 2},
    {"event_time": base + timedelta(seconds=15), "value": 6},
]
event_time_window(events, window_size_seconds=10)

class WatermarkGenerator:
    """Simulate watermark generation with bounded out-of-orderness."""

    def __init__(self, max_lateness_seconds=5):
        self.max_lateness = max_lateness_seconds
        self.max_event_time = 0
        self.current_watermark = 0

    def observe(self, event_time_epoch):
        """Observe an event and potentially advance the watermark."""
        self.max_event_time = max(self.max_event_time, event_time_epoch)
        # Watermark = max observed - max lateness
        new_watermark = self.max_event_time - self.max_lateness
        if new_watermark > self.current_watermark:
            self.current_watermark = new_watermark
        return self.current_watermark

    def should_trigger_window(self, window_end_time):
        """Should we fire this window?"""
        return self.current_watermark >= window_end_time

# Simulate
watermark = WatermarkGenerator(max_lateness_seconds=5)
events = [10, 12, 8, 14, 18, 25, 22, 30]
window_ends = [15, 20, 25, 30]

print(f"{'Event':<10} {'Observed Time':<15} {'Watermark':<12} {'Window 15':<10} "
      f"{'Window 20':<10}")
print("-" * 60)

for et in events:
    wm = watermark.observe(et)
    w15 = "⏳" if not watermark.should_trigger_window(15) else "🔥"
    w20 = "⏳" if not watermark.should_trigger_window(20) else "🔥"
    print(f"{et:<10} {et:<15} {wm:<12} {w15:<10} {w20:<10}")

class LateDataHandler:
    def __init__(self, watermark_gen, allowed_lateness=10):
        self.watermark = watermark_gen
        self.allowed_lateness = allowed_lateness
        self.late_queue = []

    def process_event(self, event):
        """Route event: normal window or late data."""
        event_time = event["event_time"]
        wm = self.watermark.observe(event_time)

        if event_time >= wm - self.allowed_lateness:
            print(f"[OnTime] Event at {event_time} processed normally")
            return "normal"
        else:
            self.late_queue.append(event)
            print(f"[Late] Event at {event_time} routed to late queue")
            return "late"

handler = LateDataHandler(WatermarkGenerator(5), allowed_lateness=10)
events_to_process = [10, 12, 20, 25, 15, 30, 8, 35]
for et in events_to_process:
    handler.process_event({"event_time": et, "value": 1})
print(f"\nLate queue size: {len(handler.late_queue)}")

class ExactlyOnceProcessor:
    """Simulate exactly-once processing with checkpointing."""

    def __init__(self):
        self.state = {}
        self._checkpoint = {}
        self._pending = {}
        self._offset = 0

    def process(self, key, value):
        """Process an event with exactly-once semantics."""
        self._pending[key] = value
        self._offset += 1

    def checkpoint(self):
        """Atomic commit: save pending state and advance offset."""
        self._checkpoint = dict(self.state)
        self.state.update(self._pending)
        self._pending = {}
        print(f"[Checkpoint] State size: {len(self.state)}, offset: {self._offset}")

    def recover_from_failure(self):
        """On failure, restore last checkpoint."""
        self.state = dict(self._checkpoint)
        self._pending = {}
        print(f"[Recovery] Restored state from checkpoint, offset reset to {self._offset}")
        # Events after the checkpoint will be replayed from source

processor = ExactlyOnceProcessor()
processor.process("a", 10)
processor.process("b", 20)
processor.checkpoint()
processor.process("a", 15)  # Update after checkpoint
processor.process("c", 30)

print("Before recovery:", processor.state)
processor.recover_from_failure()
print("After recovery:", processor.state)

# Stateless: map, filter, flatMap
def stateless_filter(event):
    """Stateless: pure function of single event."""
    return event["value"] > 0

def stateless_map(event):
    """Stateless: transform each event independently."""
    return {"user": event["user"], "page": event["page"].lower()}

events = [
    {"user": "Alice", "page": "/Home", "value": 1},
    {"user": "Bob", "page": "/About", "value": -1},
]
print("Stateless filter:", [e for e in events if stateless_filter(e)])
print("Stateless map:", [stateless_map(e) for e in events])

class StatefulCounter:
    """Stateful: maintain running count per key."""

    def __init__(self):
        self._counts = {}

    def process(self, key):
        self._counts[key] = self._counts.get(key, 0) + 1
        return self._counts[key]

    def snapshot(self):
        return dict(self._counts)

counter = StatefulCounter()
for key in ["a", "b", "a", "a", "c", "b"]:
    count = counter.process(key)
    print(f"Key {key}: count={count}")

class KappaPipeline:
    """Simulate Kappa: single stream pipeline with replay capability."""

    def __init__(self):
        self._event_log = []  # Kafka-like append-only log
        self._materialized_view = {}

    def ingest(self, events):
        """Append to immutable log."""
        for e in events:
            self._event_log.append(e)

    def run_pipeline(self, from_offset=0):
        """Process from any offset (replay/batch capability)."""
        state = {}
        for i in range(from_offset, len(self._event_log)):
            e = self._event_log[i]
            key = e["user"]
            state[key] = state.get(key, 0) + e["value"]
        self._materialized_view = state
        return state

    def query_realtime(self):
        """Query the latest materialized view."""
        return self._materialized_view

    def query_historical(self, on_date):
        """Reprocess up to a point in time."""
        return self.run_pipeline()

# Same pipeline for real-time and batch
kappa = KappaPipeline()
kappa.ingest([
    {"user": "alice", "value": 5, "ts": 1},
    {"user": "bob", "value": 3, "ts": 2},
])
print("Real-time view:", kappa.run_pipeline())

# Reprocess for correction/batch
kappa.ingest([
    {"user": "alice", "value": 2, "ts": 3},
    {"user": "charlie", "value": 7, "ts": 4},
])
print("After reprocess:", kappa.run_pipeline(from_offset=0))

# streaming_pipeline.py
# Simulate a complete streaming pipeline with time semantics
from datetime import datetime
from collections import defaultdict
import random
import time

class StreamingPipeline:
    def __init__(self, allowed_lateness_seconds=5):
        self.watermark = WatermarkGenerator(allowed_lateness_seconds)
        self.windows = defaultdict(list)
        self.results = {}
        self.metrics = {"total": 0, "late": 0, "on_time": 0, "windows_fired": 0}

    def process(self, event):
        et = event["event_time"]
        wm = self.watermark.observe(et)
        window_key = (et // 10) * 10
        self.windows[window_key].append(event)
        self.metrics["total"] += 1

        if et >= wm - self.watermark.max_lateness:
            self.metrics["on_time"] += 1
        else:
            self.metrics["late"] += 1

        # Fire completed windows
        for wk in list(self.windows.keys()):
            if wm >= wk + 10 and wk not in self.results:
                values = [e["value"] for e in self.windows[wk]]
                avg = sum(values) / len(values) if values else 0
                self.results[wk] = {"count": len(values), "avg": round(avg, 2)}
                self.metrics["windows_fired"] += 1

    def report(self):
        print(f"\nPipeline Metrics:")
        for k, v in self.metrics.items():
            print(f"  {k}: {v}")
        print(f"  Fired windows: {sorted(self.results.keys())}")

class WatermarkGenerator:
    def __init__(self, max_lateness_seconds):
        self.max_lateness = max_lateness_seconds
        self.max_event_time = 0
        self.current_watermark = 0

    def observe(self, event_time):
        self.max_event_time = max(self.max_event_time, event_time)
        self.current_watermark = self.max_event_time - self.max_lateness
        return self.current_watermark

pipeline = StreamingPipeline(allowed_lateness_seconds=3)
now = int(time.time())
for i in range(30):
    delay = random.randint(-5, 2)  # Some events arrive late
    event = {"event_time": now + i + delay, "value": random.randint(1, 100)}
    pipeline.process(event)

pipeline.report()