Accumulator

Accumulator is a special kind of window similar to a Session Window designed for complex operations like reordering, custom triggering, and joining multiple ordered streams. Like other windowing strategies (fixed, sliding, or session windows), the Accumulator window maintains state for each key, but unlike others, it allows for manipulation of the Datum and emitting them based on custom rules (e.g., sorting) . Accumulator solves is a different type of problem outside both map/flatmap (one to ~one) and reduce (many to ~one) and instead of Message, we have to emit back the "manipulated" Datum.

Another difference between the Accumulator and the Session windows is that in Accumulator, there is no concept of window merge.

Why Accumulator?

Accumulator is a powerful concept that lets you tap into the raw Datum stream and manipulate not just the order but the Datum stream itself. It has a powerful semantics where the input and output is a stream of Datum creating a Global Window. It opens up the possibility of very advanced use cases like custom triggers (e.g., count based triggers combined with windowing strategies).

Go

func Accumulator(in <-chan Datum) <-chan Datum {
    out := make(chan Datum)
    go func() {
        defer close(out)
        var state []Datum
        for i := range in {
            if WatermarkProgressed(i) {
                sort.Slice(state, func(a, b int) bool {
                    return state[a].Timestamp < state[b].Timestamp
                })
                for _, d := range state {
                    out <- d
                }
                state = nil
            }
            state = append(state, i)
        }
    }()
    return out
}

Python

def Accumulator(input_stream):
    """
    Processes an input stream of Datum objects, maintaining an ordered state.
    Emits elements when the watermark progresses.
    """
    state = OrderedList()
    output_stream = []

    for i in input_stream:
        # The condition will return True if watermark progresses
        if WatermarkProgressed(i):
            # Pop all sorted elements and write to output stream
            output_stream.extend(state.pop_n())

        # Insert the current element into the ordered state
        state.insert(i)

    return output_stream

Rust

struct Accumulator {
    state: OrderedList<Datum>,
}

impl Accumulator {
    fn new() -> Self {
        Self {
            state: OrderedList::new(),
        }
    }

    fn process(&mut self, input_stream: &[Datum], output_stream: &mut Vec<Datum>) {
        for i in input_stream {
            // Check if the watermark has progressed
            if WatermarkProgressed(i) {
                // Pop all sorted elements and write to output stream
                let popped = self.state.pop_all();
                output_stream.extend(popped);
            }
            self.state.insert(i.clone());
        }
    }
}

Java

import java.util.ArrayList;
import java.util.List;

public class Accumulator {
    private OrderedList<Datum> state;

    public Accumulator() {
        state = new OrderedList<>();
    }

    public void process(List<Datum> inputStream, List<Datum> outputStream) {
        for (Datum i : inputStream) {
            // Check if the watermark has progressed
            if (WatermarkProgressed(i)) {
                // Pop all sorted elements and write to output stream
                List<Datum> popped = state.popAll();
                outputStream.addAll(popped);
            }
            state.insert(i);
        }
    }
}

Considerations

The Accumulator window is powerful but should be used carefully as it can cause pipeline stalling if not configured properly.

Factors to consider

Please consider the following factors when using the Accumulator window (not comprehensive):

1. For high-throughput scenarios, ensure adequate storage is provisioned
2. The timeout should be set based on the expected data arrival patterns and latency requirements
3. Consider the trade-off between data completeness (longer timeout) and processing latency (shorter timeout)
4. Please make sure Watermark is honored when publishing the data, else completeness and correctness is not guaranteed

Data Retention

To ensure there is no data loss during pod restarts, the Accumulator window replays data from persistent storage. The system stores data until Outbound(Watermark) - 1, which means it keeps the minimum necessary data to ensure correctness while managing resource usage.

Constraints

1. For data older than Outbound(Watermark) - 1, users need to bring in an external store and implement replay on restart
2. Data deletion is based on the Outbound(Watermark)

Few general use cases

1. Stream Joining: Combining multiple ordered streams into a single ordered output
2. Event Reordering: Handling out-of-order events and ensuring they're processed in the correct sequence
3. Time-based Correlation: Correlating events from different sources based on their timestamps
4. Custom Sorting: Implementing user-defined sorting logic for event streams
5. Custom Triggering: Triggering actions based on specific conditions or events within the stream

Configuration

vertices:
  - name: my-udf
    udf:
      groupBy:
        window:
          accumulator:
            timeout: duration

NOTE: A duration string is a possibly signed sequence of decimal numbers, each with optional fraction and a unit suffix, such as "300ms", "1.5h" or "2h45m". Valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h".

timeout

The timeout is the duration of inactivity (no data flowing in for a particular key) after which the accumulator state is removed. This helps prevent memory leaks by cleaning up state for keys that are no longer active.

Note: The determination of whether a key is inactive, or has timed out, is based on the watermark progressing. In order to close the accumulator window, we compare this timeout against the watermark. If the watermark progression for the vertex has stalled for some reason, eg: due to one of the sources idling in a multi-source setup, the timeout may not be triggered without configuring idle watermark detection. Currently, in such cases, the accumulator window may not close as it continues to hold on to the state for the key while ingesting more data, hoping to progress watermark with the next datum. This might lead to OOM situations.

How It Works

The Accumulator maintains a single, per-key global window. The window opens when the first Datum for a key arrives and, unlike fixed, sliding, or session windows, it has no fixed end time — it stays open, accumulating (and optionally re-emitting) Datums, until we don't receive any more similar keyed events for timeout duration.

For each key, the user-defined function typically:

1. Maintains an ordered list of elements for each key (for example, sorted by event time)
2. When the watermark progresses, pops all elements that are now behind the watermark and writes them to the output stream
3. Inserts newly arrived elements into the ordered list based on their event time
4. Flushes any remaining buffered elements when the window is closed (see How windows are closed)

Unlike both map or reduce operations, where Datum is consumed and Message is returned, for reordering with the Accumulator, the Datum is kept intact.

How windows are closed (internal EOF)

Accumulator windows are closed by numaflow-core, not by the user-defined function (similar to other reduce operations). A window is closed once the watermark advances past the latest event time seen for that key plus the configured timeout — in other words, once the key has been inactive for timeout, as measured against the watermark. (As described under timeout, this is why the watermark must keep progressing: a stalled watermark can keep a window open indefinitely and lead to OOM.)

When core closes a window, it sends an internal close / end-of-stream (EOF) signal to the UDF for that key. This signal surfaces differently in each SDK, but it always means the same thing: no more input will arrive for this window, so flush whatever you have buffered.

After the UDF has handled this signal and returned, the SDK sends an EOF response back to core for that window. This EOF response is what tells core the window is fully processed, so that it can garbage-collect the window's persisted state (WAL), release the tracked messages, and let the watermark advance downstream. Data is retained until Outbound(Watermark) - 1 (see Data Retention).

Dropping or skipping messages

Because the Accumulator is a reduce-family operation — a stream of Datum in, a stream of Datum out — there is no requirement for a one-to-one mapping between input and output. If you do not want to forward a particular Datum, simply do not emit it; you do not need to emit an explicit "drop" message to account for it.

Behavior change in v1.8.1

Before v1.8.1, an accumulator window's persisted state was cleaned up only once the UDF emitted at least one output message for that window. A UDF that filtered out (dropped) every message for a window would leave that window's state uncollected, causing unbounded memory and WAL growth. This is why an explicit drop helper (for example, MessageToDrop / to_drop()) was previously considered necessary — to "account for" every input.

As of v1.8.1 (numaflow#3461), window cleanup is driven by the internal EOF described above and no longer depends on the UDF emitting output. You can safely skip messages by simply not emitting them; the window and its state are cleaned up automatically once the window closes after the timeout. Explicitly emitting drops is no longer required. (Some SDKs, such as Python, still expose a to_drop() helper, but it is optional and unnecessary for cleanup.)

Example

Here's an example of using an Accumulator window to join and sort two HTTP sources:

apiVersion: numaflow.numaproj.io/v1alpha1
kind: Pipeline
metadata:
  name: simple-accumulator
spec:
  vertices:
    - name: http-one
      scale:
        min: 1
        max: 1
      source:
        http: {}
    - name: http-two
      scale:
        min: 1
        max: 1
      source:
        http: {}
    - name: accum
      udf:
        container:
          # stream sorter example
          image: quay.io/numaio/numaflow-go/stream-sorter:stable
        groupBy:
          window:
            accumulator:
              timeout: 10s
          keyed: true
          storage:
            persistentVolumeClaim:
              volumeSize: 1Gi
    - name: sink
      scale:
        min: 1
        max: 1
      sink:
        log: {}
  edges:
    - from: http-one
      to: accum
    - from: http-two
      to: accum
    - from: accum
      to: sink

In this example:

1. We have two HTTP sources (http-one and http-two) that produce ordered streams
2. The accum vertex uses an Accumulator window with a timeout of 10 seconds
3. The accumulator joins and sorts the events from both sources based on their event time
4. The sorted output is sent to a log sink

Note: Setting readBatchSize: 1 helps maintain the ordering of events in the input streams.

Check out the snippets below to see the UDF examples for different languages:

Go

func (s *streamSorter) Accumulate(ctx context.Context, input <-chan accumulator.Datum, output chan<- accumulator.Message) {
    for {
        select {
        case <-ctx.Done():
            log.Println("Exiting the Accumulator")
            return
        case datum, ok := <-input:
            // this case happens due to timeout
            if !ok {
                log.Println("Input channel closed")
                return
            }
            log.Println("Received datum with event time: ", datum.EventTime().UnixMilli())

            // watermark has moved, let's flush
            if datum.Watermark().After(s.latestWm) {
                s.latestWm = datum.Watermark()
                s.flushBuffer(output)
            }

            // store the data into the internal buffer
            s.insertSorted(datum)
        }
    }
}

View the Full Example on numaflow-go Github

Python

class StreamSorter(Accumulator):
  def __init__(self):
      _LOGGER.info("StreamSorter initialized")
      self.latest_wm = datetime.fromtimestamp(-1)
      self.sorted_buffer: list[Datum] = []

  async def handler(
      self,
      datums: AsyncIterable[Datum],
      output: NonBlockingIterator,
  ):
      _LOGGER.info("StreamSorter handler started")
      async for datum in datums:
          _LOGGER.info(
              f"Received datum with event time: {datum.event_time}, "
              f"Current latest watermark: {self.latest_wm}, "
              f"Datum watermark: {datum.watermark}"
          )

          # If watermark has moved forward
          if datum.watermark and datum.watermark > self.latest_wm:
              self.latest_wm = datum.watermark
              _LOGGER.info(f"Watermark updated: {self.latest_wm}")
              await self.flush_buffer(output)

          self.insert_sorted(datum)

      _LOGGER.info("Timeout reached")
      await self.flush_buffer(output, flush_all=True)

View the full example on numaflow-python Github

Rust

  /// insert_sorted will do a binary-search and inserts the AccumulatorRequest into the sorted buffer.
  fn insert_sorted(sorted_buffer: &mut Vec<AccumulatorRequest>, request: AccumulatorRequest) {
      let event_time = request.event_time;

      // Find the insertion point using binary search
      let index = sorted_buffer
          .binary_search_by(|probe| probe.event_time.cmp(&event_time))
          .unwrap_or_else(|e| e);

      sorted_buffer.insert(index, request);
  }

View the full example in nummaflow-rs on Github

Java

  @Slf4j
  @AllArgsConstructor
  public class StreamSorterFactory extends AccumulatorFactory<StreamSorterFactory.StreamSorter> {

      public static void main(String[] args) throws Exception {
          log.info("Starting stream sorter server..");
          Server server = new Server(new StreamSorterFactory());

          // Start the server
          server.start();

          // wait for the server to shut down
          server.awaitTermination();
          log.info("Stream sorter server exited..");
      }

      @Override
      public StreamSorter createAccumulator() {
          return new StreamSorter();
      }

      public static class StreamSorter extends Accumulator {
          private Instant latestWm = Instant.ofEpochMilli(-1);
          private final TreeSet<Datum> sortedBuffer = new TreeSet<>(Comparator
                  .comparing(Datum::getEventTime)
                  .thenComparing(Datum::getID)); // Assuming Datum has a getUniqueId() method

          @Override
          public void processMessage(Datum datum, OutputStreamObserver outputStream) {
              log.info("Received datum with event time: {}", datum.toString());
              if (datum.getWatermark().isAfter(latestWm)) {
                  latestWm = datum.getWatermark();
                  flushBuffer(outputStream);
              }
              sortedBuffer.add(datum);
          }

          @Override
          public void handleEndOfStream(OutputStreamObserver outputStreamObserver) {
              log.info("Eof received, flushing sortedBuffer: {}", latestWm.toEpochMilli());
              flushBuffer(outputStreamObserver);
          }

          private void flushBuffer(OutputStreamObserver outputStream) {
              log.info("Watermark updated, flushing sortedBuffer: {}", latestWm.toEpochMilli());
              while (!sortedBuffer.isEmpty() && sortedBuffer
                      .first()
                      .getEventTime()
                      .isBefore(latestWm)) {
                  Datum datum = sortedBuffer.pollFirst();
                  assert datum != null;
                  outputStream.send(new Message(datum));
                  log.info("Sent datum with event time: {}", datum.getEventTime().toEpochMilli());
              }
          }
      }
  }

View the full example in numaflow-java Github