Namespace reduceStream

Reduce Stream namespace for streaming aggregation results.

Similar to Reduce, but outputs results as an async iterable stream instead of an array. This is useful when:

Aggregation produces incremental results
You want to emit partial aggregates as data arrives
Output size is unknown or potentially large

Example

import { reduceStream } from '@numaproj/numaflow-js';

const server = new reduceStream.AsyncServer(async function* (keys, datums, metadata) {
  let runningSum = 0;

  for await (const datum of datums) {
    runningSum += parseInt(datum.value.toString());

    // Emit running total
    yield new reduceStream.Message(Buffer.from(JSON.stringify({
      keys,
      runningSum,
      window: metadata.intervalWindow
    })));
  }
});

server.start();

Classes

AsyncServer
Message

Interfaces

Datum
IntervalWindow
MessageOptions
Metadata

Type Aliases

CallbackFn

Namespace reduceStream

Example

Classes

Interfaces

Type Aliases

Settings

On This Page