How to Implement a Streaming Watermark in Python

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_event(self, event_time):
        """Track the latest event timestamp seen."""
        if self.max_event_time is None or event_time > self.max_event_time:
            self.max_event_time = event_time

    def get_watermark(self, current_time):
        """Compute watermark = max_event_time - delay."""
        if self.max_event_time is None:
            return None
        return self.max_event_time - self.watermark_delay


def simulate_stream(events, watermark_delay_seconds=10):
    """Simulate a stream and show how the watermark evolves."""
    tracker = StreamingWatermark(watermark_delay_seconds)
    print(f"Watermark delay: {watermark_delay_seconds}s")
    print("=" * 50)

    current_time = datetime(2024, 1, 1, 12, 0, 0)
    for event_time in events:
        tracker.observe_event(event_time)
        current_time += timedelta(seconds=3)  # simulate clock ticking
        watermark = tracker.get_watermark(current_time)

        print(f"Event: {event_time.strftime('%H:%M:%S')} | "
              f"MaxEvent: {tracker.max_event_time.strftime('%H:%M:%S')} | "
              f"Watermark: {watermark.strftime('%H:%M:%S')}")


if __name__ == "__main__":
    # Events arriving with some out-of-order data
    base = datetime(2024, 1, 1, 12, 0, 0)
    events = [
        base + timedelta(seconds=0),
        base + timedelta(seconds=5),
        base + timedelta(seconds=3),   # out of order
        base + timedelta(seconds=12),
        base + timedelta(seconds=8),   # out of order
        base + timedelta(seconds=20),
    ]

    simulate_stream(events, watermark_delay_seconds=10)