Stream-based Real-time Aggregation

In this lesson, you will learn how to maintain memory efficiency while calculating real-time statistics—like moving averages—on high-velocity streaming data.

The Problem with Naive Aggregation

If you collect all incoming stream data into an array to calculate an average, you will eventually hit an OutOfMemory error as your data grows. To build a "sliding window" (e.g., the average value of events over the last 60 seconds), we need a strategy that discards old data while keeping only what is necessary for the current calculation.

Building a Low-Memory Aggregator

Instead of storing every raw event, we store a small state object. We use a Transform stream to process data as it arrives. For a sliding window, we keep a queue of timestamped values and periodically prune entries that fall outside our desired time window.

const { Transform } = require('stream');

class MovingAverage extends Transform {
  constructor(windowSizeMs) {
    super({ objectMode: true });
    this.windowSizeMs = windowSizeMs;
    this.buffer = []; // Stores { timestamp, value }
    this.sum = 0;
  }

  _transform(chunk, encoding, callback) {
    const now = Date.now();
    this.buffer.push({ timestamp: now, value: chunk.value });
    this.sum += chunk.value;

    // Prune old data
    while (this.buffer.length > 0 && (now - this.buffer[0].timestamp) > this.windowSizeMs) {
      this.sum -= this.buffer.shift().value;
    }

    this.push({ average: this.sum / this.buffer.length });
    callback();
  }
}

Why this is memory-efficient

By using a Transform stream, we process each chunk exactly once. Because we are shift()-ing out values that are older than our windowSizeMs, the buffer never grows beyond the number of events that occur within that specific time slice. If your stream receives 100 events per second and your window is 1 minute, your memory usage remains constant regardless of whether the stream runs for an hour or a year.

By managing state locally within the stream pipeline, you bypass the need for external storage or bulk arrays, keeping your application footprint small even under high-load telemetry scenarios.

Key Takeaway: Real-time aggregation should be handled by discarding expired state within the stream's buffer, ensuring memory consumption remains constant regardless of total stream duration.