StreamTable

A stream is a split table whose rows keep arriving. Each worker parses its own batches (256 KB at a time, one batch per event-loop turn, so a query waits for at most one batch) into growing columns, keeps the newest rows up to a limit, and answers queries over the rows it holds. Rows reach the workers either through push() on this thread or through connect(), where each worker reads its own TCP or WebSocket feed and the data never passes through this thread. A LiveWindow keeps small running totals per time slice in every worker, so reading “the last five minutes per region” costs about a millisecond. Streams are experimental: the API may change. See Streams and live windows.

import { SplitEngine } from 'vectrojs';           // Node
import { SplitEngine } from 'vectrojs/browser';   // browsers
const live = await new SplitEngine().streamAsync({ columns: { ts: 'number', region: 'category' } });

Opening a stream

engine.streamAsync(options)

A method of SplitEngine. Returns a promise of a StreamTable.

Name	Type	Default	Meaning
columns	object	required	{ name: 'number' or 'category' or 'date' }. Dates are milliseconds, or ISO text in CSV and JSON lines.
format	string	'csv'	'csv' (lines without a header, fields in the order of columns), 'ndjson' (one flat JSON object per line; columns picks the keys), 'vectro' (binary frames from an Emitter) or 'arrow' (an Arrow IPC stream per connection).
keepRows	number	Infinity	Rows kept in total, spread over the workers; the oldest are dropped.
keepMs	number	none	With timeColumn: keep rows newer than the newest time minus keepMs.
timeColumn	string	none	The time column keepMs measures.

JSON lines are parsed byte by byte straight into columns: keys may come in any order, extra and nested keys are skipped, null and missing values are empty, and lines that aren’t JSON objects are counted (badLines) and skipped. Streams take flat records only.

Feeding it

stream.push(chunk)

For 'csv' and 'ndjson': a string or bytes cut anywhere. Whole lines go to the workers in turn; an unfinished last line waits for the next push. Returns a promise that settles when the worker has queued the lines. When a worker has more than 4 MB waiting to be parsed, it delays its answer until it’s down to 1 MB, so awaiting push slows a fast producer. Binary formats reject push; connect them instead.

stream.connect(worker, options)

Has worker number worker open its own connection and read from it: { url: 'wss://...' } for a WebSocket (browsers, Node 22 and later; each message holds whole lines or one binary frame), or { host, port } for TCP (Node; host defaults to '127.0.0.1'). Returns a promise that settles once the connection is open. A TCP socket is paused while its worker is backed up, which slows the sender; a WebSocket can’t be paused.

await Promise.all(engine.workers.map((_, w) => stream.connect(w, { host: 'feed.local', port: 9000 })));

stream.flushAsync()

Sends any unfinished last line (with a newline added) and has every worker parse what it has received. Returns the same totals as statsAsync. Queries see the rows parsed so far, so call it first when you need every row pushed.

stream.statsAsync()

{ rows, rowsIn, badLines, bytesIn, batches, unparsedBytes, perWorker, errors }: rows kept now, rows taken in, lines skipped, bytes received, batches parsed, bytes not parsed yet, rows kept per worker, and connection or parse errors.

Querying it

A stream answers the split-table queries over the rows its workers hold right now: aggAsync, sumAsync, meanAsync, minAsync, maxAsync, countAsync, groupBy(keys).aggAsync, topAsync, toArrowAsync and dispose, as documented on SplitTable. Group keys sort A to Z, since the workers have no shared dictionary. These differ:

stream.length

Always NaN: the number changes all the time. Use countAsync().

stream.filter(expr, params?)

A filtered view of the stream, still live.

stream.snapshotAsync()

A fixed copy of the rows kept now (those matching the filters), as a normal split table that deriveAsync, collectAsync and the rest work on. Also collectAsync(), which does the same.

stream.toRowsAsync(limit?)

The rows kept now as plain objects. Rows have no order across workers: worker 0’s rows come first.

stream.toColumnsAsync()

The rows kept now as plain columns (through a snapshot that is disposed afterwards).

stream.deriveAsync()

Throws: take a snapshotAsync() first.

Live windows

stream.windowAsync(options)

Starts running totals over a time window and returns a promise of a LiveWindow. Every worker keeps one small total per (slice of every milliseconds, group), updated as rows arrive; totals outlive rows dropped by keepRows. The window ends at the newest time seen and covers whole slices.

Name	Type	Default	Meaning
agg	object	required	Aggregations as in table.agg. Use approxDistinct for distinct counts (sketches per slice, within about 1.6%); countDistinct is refused.
every	number	required	Slice length in milliseconds.
last	number	required	Window length in milliseconds.
timeColumn	string	required	The column the slices are cut by.
by	string	none	One category column to group by.

window.readAsync(options?)

Reads the window: one row per group (or one row without by), with the agg names as columns. With { series: true }, one row per slice and group, with t, the slice’s start in milliseconds, for charts. Rows come sorted by t, then by group label.

window.dispose()

Stops the totals on every worker. Returns a promise.

import { SplitEngine } from 'vectrojs';

const engine = new SplitEngine({ workers: 2 });
const orders = await engine.streamAsync({
  format: 'csv',
  columns: { ts: 'number', region: 'category', price: 'number' },
  keepRows: 10_000,
});
const live = await orders.windowAsync({
  by: 'region', every: 1000, last: 60_000, timeColumn: 'ts',
  agg: { orders: 'count()', revenue: 'sum(price)', prices: 'approxDistinct(price)' },
});

await orders.push('1000,EU,10\n1500,US,20\n2200,EU,');   // cut anywhere
await orders.push('30\n3100,EU,5\n');
console.log(await orders.flushAsync());

console.log(await live.readAsync());
console.log(await live.readAsync({ series: true }));
console.log(await orders.filter("region == 'EU'").sumAsync('price'), await orders.countAsync());

const fixed = await orders.snapshotAsync();
const taxed = await fixed.deriveAsync('gross', 'price * 1.2');
console.log((await taxed.topAsync(1, '-gross')).toRows());

await live.dispose();
await engine.close();

import { SplitEngine } from 'vectrojs';

const engine = new SplitEngine({ workers: 2 });
const events = await engine.streamAsync({
  format: 'ndjson',
  columns: { ts: 'date', user: 'category', ms: 'number' },
  keepMs: 60_000, timeColumn: 'ts',
});

await events.push([
  '{"ts":"2024-06-01T12:00:00Z","user":"ann","ms":120}',
  '{"user":"bob","ms":80,"ts":"2024-06-01T12:00:20Z","extra":true}',
  'not json',
  '{"ts":"2024-06-01T12:00:40Z","user":"ann","ms":null}',
].join('\n') + '\n');

const stats = await events.flushAsync();
console.log(stats.rows, stats.badLines);
console.log((await events.groupBy('user').aggAsync({ n: 'count()', avg: 'mean(ms)' })).toRows());
console.log(await events.toRowsAsync());
await engine.close();