When you first start working with stream processing in Scala, it's tempting to treat a stream like a very long, lazy list. You think, "I just need to calculate the average number of requests per minute," and your instinct is to maintain a stateful variable—maybe a List or a Map—that collects incoming events. I've seen this happen in plenty of PRs: a developer creates a buffer, pushes events into it, and then filters that buffer for timestamps within the last 60 seconds whenever they need a result.

The Memory Leak waiting to happen

Let's look at how that naive approach usually looks in code. Imagine we're tracking "Add to Cart" events for an e-commerce site to detect spikes in popularity.

case class CartEvent(productId: String, timestamp: Long)

var eventBuffer = List.empty[CartEvent]

def processEvent(event: CartEvent): Int = {
  eventBuffer = event :: eventBuffer
  val oneMinuteAgo = event.timestamp - 60000
  
  // Filter out the old stuff every time we process
  eventBuffer = eventBuffer.filter(_.timestamp > oneMinuteAgo)
  
  eventBuffer.size
}

On the surface, this works. You're filtering out the old data, so why is this "wrong"? The problem is the cost of the filter operation. Every single time a new event arrives, you are iterating over the entire last minute of data. If you're processing 10,000 events per second, you're performing 10,000 linear scans per second on a list that could contain 600,000 elements. You'll spend more CPU cycles traversing pointers in a linked list than actually doing business logic. Even worse, because you're creating a new list on every update, you're putting an immense amount of pressure on the JVM garbage collector. I've watched this pattern bring down a production cluster because the GC pauses became longer than the processing window itself.

Tumbling Windows and the Bucket Strategy

The professional way to handle this is through windowing. Instead of treating the stream as a continuous line, we chop it into discrete "buckets" of time. In a tumbling window, these buckets are fixed and non-overlapping. If you want a one-minute window, you group everything from 10:00 to 10:01, then 10:01 to 10:02.

By mapping the event timestamp to a bucket ID—essentially timestamp / windowSize—you can use a Map where the key is the window start time and the value is the aggregated total. You aren't filtering a list; you're just incrementing a counter in a specific bucket. This changes the complexity from O(n) per event to effectively O(1).

case class WindowKey(startTime: Long)

// Using a map to track aggregates per window
var windows = Map.empty[WindowKey, Int]
val windowSize = 60000L

def processEventBetter(event: CartEvent): Int = {
  val bucketStart = (event.timestamp / windowSize) * windowSize
  val key = WindowKey(bucketStart)
  
  val newCount = windows.getOrElse(key, 0) + 1
  windows = windows + (key -> newCount)
  
  // Periodically purge windows that are truly ancient
  if (windows.size > 10) { 
    val cutoff = System.currentTimeMillis() - (windowSize * 5)
    windows = windows.filter { case (k, _) => k.startTime > cutoff }
  }
  
  newCount
}

Sliding Windows: Balancing Precision and Performance

Tumbling windows are great, but they have a "cliff" problem. If a spike in traffic happens right at the boundary of two windows, the spike is split in half, and your alert might not trigger. To fix this, we use sliding windows. A sliding window allows you to say, "Give me the count for the last 60 seconds, updated every 10 seconds."

The trade-off here is state duplication. Since a single event now falls into multiple overlapping windows, you have to update multiple buckets for every single event. If your window is 60 seconds and your slide is 10 seconds, each event belongs to 6 different windows. You're trading a bit of memory and CPU for much smoother, more accurate trend lines. When I'm designing these systems, I always ask: "Does the business need to know the exact second the spike started, or is a 10-second lag acceptable?" Usually, the lag is fine, and the performance gain is massive.

📋 Practical Task

Implementation: Real-time Temperature Moving Average

You are building a monitoring system for a server room. You receive a stream of temperature readings, and you need to calculate a sliding average of the temperature over the last 30 seconds, updated every 5 seconds.

Implement a class TemperatureMonitor with the following requirements:

• A method addReading(temp: Double, timestamp: Long): Double that returns the current average for the most recent window.
• The system must use a "bucketed" approach (sliding window) rather than storing every single single reading in a raw list to ensure it remains performant under high load.
• Implement a mechanism to prune old buckets from your state so the memory usage doesn't grow indefinitely.

Constraints: Avoid using external streaming libraries like Akka Streams or Flink for this exercise; implement the windowing logic using standard Scala collections to demonstrate your understanding of the underlying state management.