Implement Sliding Windows - The MapReducer

                      TASK
                    

Implementation

Tumbling windows are non-overlapping — an event belongs to exactly one window. Sliding windows overlap: each event belongs to multiple windows, enabling smooth rolling aggregations like "average latency over the last 5 minutes, updated every minute".

window_size=5min, slide=1min

Events at 10:02: ---e1---

Windows containing e1:
  [10:00 - 10:05]  <- window starting at 10:00
  [10:01 - 10:06]  <- window starting at 10:01
  [10:02 - 10:07]  <- window starting at 10:02

Your node handles two message types:

// Assign one event to all sliding windows it belongs to
{ "type": "assign", "msg_id": 1,
  "event_timestamp": "2024-01-15T10:02:00Z",
  "current_time":    "2024-01-15T10:03:00Z",
  "window_size_ms":  300000,
  "slide_ms":        60000 }
→ { "type": "assigned", "in_reply_to": 1,
    "windows": [
      {"window_id": "...", "window_start": "2024-01-15T10:00:00Z", "window_end": "2024-01-15T10:05:00Z"},
      {"window_id": "...", "window_start": "2024-01-15T10:01:00Z", "window_end": "2024-01-15T10:06:00Z"},
      {"window_id": "...", "window_start": "2024-01-15T10:02:00Z", "window_end": "2024-01-15T10:07:00Z"}
    ]}

// List all windows active at the given time
{ "type": "active_windows", "msg_id": 2,
  "current_time": "2024-01-15T10:03:00Z",
  "window_size_ms": 300000,
  "slide_ms": 60000 }
→ { "type": "active_windows_result", "in_reply_to": 2, "count": 5 }

The key difference from tumbling windows: window_size_ms / slide_ms windows are active at any point in time.

Sample Test Cases

Assign to sliding windowsTimeout: 5000ms

Input

{
  "src": "stream",
  "dest": "windower",
  "body": {
    "type": "assign",
    "msg_id": 1,
    "event_timestamp": "2024-01-15T10:02:00Z",
    "current_time": "2024-01-15T10:03:00Z",
    "window_size_ms": 300000,
    "slide_ms": 60000
  }
}

Expected Output

{"type": "assigned", "in_reply_to": 1, "windows": [{"window_id": "window-1705305600000-1705307400000", "window_start": "2024-01-15T10:00:00Z", "window_end": "2024-01-15T10:05:00Z"}, {"window_id": "window-1705305660000-1705307460000", "window_start": "2024-01-15T10:01:00Z", "window_end": "2024-01-15T10:06:00Z"}, {"window_id": "window-1705305720000-1705307520000", "window_start": "2024-01-15T10:02:00Z", "window_end": "2024-01-15T10:07:00Z"}]}

Count active windowsTimeout: 5000ms

Input

{
  "src": "stream",
  "dest": "windower",
  "body": {
    "type": "active_windows",
    "msg_id": 1,
    "current_time": "2024-01-15T10:03:00Z",
    "window_size_ms": 300000,
    "slide_ms": 60000
  }
}

Expected Output

{"type": "active_windows_result", "in_reply_to": 1, "count": 5}

Hints

Hint 1▾

An event at time T belongs to every window whose [start, end) contains T

Hint 2▾

Window starts: all multiples of slide_ms from (T - window_size_ms) up to T

Hint 3▾

Number of windows per event = window_size_ms / slide_ms

Hint 4▾

For active_windows, list all windows that overlap with current_time

Hint 5▾

Moving average: sum event values across the window, divide by count

Resources

Streaming 102 — The World Beyond Batch

Windowing models: tumbling, sliding, and session windows

                      OVERVIEW
                    

Theoretical Hub

Concept overview coming soon

Key Concepts

sliding windowsoverlapping windowswindow sizeslide intervalmoving average

main.py

python