@gestaltbi/stream

Documentation →

Framework-agnostic streaming data-processing pipeline. Compose named ops over RxJS observables with JSON-defined process graphs.

This package powers the GestaltBI client and is intended for use behind a rete.js visual editor for authoring processing.json graphs interactively.

Once published to npm:

npm install @gestaltbi/stream rxjs olap-cube-js moment

In the meantime — direct from git, pinned to a tag:

npm install GestaltBI/stream#v0.1.0 rxjs olap-cube-js moment

The package's prepare script runs tsc after install, so a fresh dist/ is built on the consumer's machine.

rxjs, olap-cube-js and moment are peer dependencies — bring your own versions.

• Op — a unit of computation, implementing run(df) and optionally getExternal() (for ops that need to load external resources before running). The built-in ops:

  op	what it does
  clear	drop empty rows
  format	parse dates and numbers by tag
  globalfilter / localfilter	filter state, shared or per-stream
  enhance	derived columns from Polish-notation expressions, cumsum
  geocode / geojsonify / regionify / heatmap	spatial
  aggregate	OLAP roll-up — sum, avg, min, max, median, first, last, concat, ratio
  diffcalc	period-over-period deltas; a stock's first difference is its flow
  recognize	revenue recognition — spread an amount across the periods it serves
  cohort	cohort axis, plus the incomplete-window guard
  assert	run checks, return verdicts
  pivot	cross-tabulate: dimensions down the side, dimensions across the top
  correlate	how strongly columns move together — Pearson/Spearman, eta, Cramér's V
  join	bring one frame's columns onto another's rows, matched on a key
  union	stack several frames, optionally recording which branch each row came from

• OpRegistry — maps op names to op classes for dynamic instantiation. Pre-populated with the built-in ops; extend with your own.

• Processor — orchestrates streaming data through a graph of named ops. Holds the input dataframe, an OLAP cube derived from it, identifier-keyed filter state, and one Observable per active stream.

• ColumnDirectory — interface the host implements to expose column metadata (which columns are tagged "date", "currency", "geocodable", etc). The Angular adapter wraps a service that loads assets/structure.json; StructureDirectory in this package is a dependency-free reference implementation over the same document, so the pipeline runs standalone.

• Check — a predicate over a processed frame that returns a Verdict (pass / fail / warn / skip) naming the periods that broke it. See Validation.

• ExternalFetcher — (url) => Observable<any> to let ops like geocode load lookup data without depending on a specific HTTP client.

import { Processor, type ColumnDirectory, type ProcessConfig } from '@gestaltbi/stream';
import { firstValueFrom } from 'rxjs';

const columnDirectory: ColumnDirectory = {
  getColumnsFor: (tag) => /* … */,
  getDataStructureFor: (tag) => /* … */,
  getDimensionHierarchies: () => /* … */,
};

const processes: ProcessConfig = JSON.parse(/* read processing.json */);

const proc = new Processor({ columnDirectory, processes });
proc.workOn({ data: rows });

const result$ = proc.getProcessed('clean_and_aggregate');
const rows = await firstValueFrom(result$);

Ops never hard-code column names — they ask the directory for columns carrying a tag. Alongside the existing uatu:dimension / uatu:measure / uatu:dimension:time / uatu:dimension:geo / uatu:aggregable vocabulary:

All additive: untagged numeric columns keep behaving exactly as before.

assert is normally a terminal node: it returns Verdict[] instead of rows, so a host can render pass/fail cards beside the charts and re-run them on every refresh.

{ "op": "assert", "require": ["margins"], "options": { "orderBy": "month", "checks": [
  { "id": "gross-margin-positive", "type": "sign", "measure": "margin_accrual", "expect": ">0" },
  { "id": "storage-is-a-ratchet", "type": "monotonic", "measure": "s3_tib" }
] } }

window_complete measures from the end of the cohort bucket when you pass span: someone who signed up on 31 October has been observed for six days on 6 November, not thirty-six. Without that guard a half-observed cohort reports a perfect retention rate.

Checks are plain data — runCheck / runChecks are exported if you'd rather call them outside a graph.

Most of the package reads a frame along time — deltas, recognition, cohorts. pivot and correlate read it across the dimensions instead, which is what it takes to see how the things you collected relate to each other.

aggregate rolls a frame up along one axis. pivot puts a second axis across the top:

{ "op": "pivot", "require": ["clean"], "options": {
    "rows": ["product_family"],
    "columns": ["region"],
    "measure": "revenue",
    "type": "sum",
    "totals": true } }

Emits one plain record per row key — the row dimensions, then one field per column bucket — so a grid or a chart consumes it directly. rows defaults to every uatu:dimension not spent on the column axis; drop columns entirely and it degenerates to a group-by.

Cells take the same aggregation kinds as aggregate (sum, avg, median, ratio, …) plus count and countDistinct, and they fold the same way — the accumulators are shared, so a sum in a pivot cell means what a sum in a rolled-up row means. Totals re-fold the accumulators rather than the finished cells: an average of averages is not the average.

A combination with no rows behind it is null, not 0 — an absence is not a zero. Column buckets are capped (columnLimit, default 50); the tail lands under Other and is counted in getOmitted(), never dropped quietly.

Which pairs actually travel together, and how strongly. "Correlation" means a different statistic depending on what is being related, so the op picks one:

{ "op": "correlate", "require": ["clean"], "options": {
    "method": "spearman", "minCoefficient": 0.3, "limit": 20 } }

Terminal, like assert: it returns Association records rather than rows, strongest first, each with a one-line summary safe to render straight into a card. Pairs it cannot honestly score — a constant column, too few complete rows, a dimension with more levels than maxLevels — come back with a null coefficient and a reason instead of a number.

A coefficient is a description, not a cause. Two measures derived from one another — a total and its own component — will sit near 1 and mean nothing.

processing.json describes a DAG of named processes:

{
  "process": {
    "format": {
      "op": "format",
      "options": { "dateTag": "uatu:date", "dateFormat": "DD/MM/YYYY" }
    },
    "clean_dates": {
      "op": "clear",
      "require": ["format"]
    }
  }
}

require is a dataflow edge: a process reads the output of what it requires, and nothing else.

Fan-out is free. A stage that several branches read is built once and, because each node's stream is shared, runs once per emission however many consumers it has — so a graph like this evaluates clean a single time:

{
  "clean":    { "op": "clear",     "require": ["format"] },
  "by_month": { "op": "aggregate", "require": ["clean"], "options": { "groupby": ["uatu:date"] } },
  "by_region":{ "op": "aggregate", "require": ["clean"], "options": { "groupby": ["region"] } }
}

Fan-in needs an op that asks for it. require may name several processes, but an op only receives more than one frame if it declares that it reads more than one — see Ops that combine frames. Wiring two inputs into an op that reads one is an error rather than a silently discarded branch, and a cycle is reported by name instead of overflowing the stack.

Each process is resolved per stream identifier, so two views reading the same graph through different filters never share a cached frame.

join and union read two or more processes:

{
  "actuals":  { "op": "aggregate", "require": ["clean"], "options": { "groupby": ["sku"] } },
  "targets":  { "op": "clear",     "require": ["target_import"] },
  "vs_target":{ "op": "join", "require": ["actuals", "targets"],
                "options": { "on": "sku", "prefix": "target:" } }
}

The frames arrive in the order require names them. join is a left join by default — enrichment should not drop the rows that failed to match — and takes only the first match on a duplicated key, because a join that quietly multiplies rows breaks every sum below it. union stacks frames and can record which branch each row came from, which is usually the point: it turns the origin into a dimension you can group by.

To write your own, override runAll instead of run and declare the arity:

export class Interleave extends AbstractOp {
  public override readonly inputs = 2;
  public override runAll([left, right]) {
    return left.flatMap((row, i) => (right[i] ? [row, right[i]] : [row]));
  }
}

An op that does not implement runAll keeps the original contract untouched: df[0] is the upstream frame, df[1] the external resources.

An end-to-end fixture built from everpix/Everpix-Intelligence, the dataset the Everpix team published when the company shut down in 2013. It is a good test subject because it publishes both sides of the recognition question: monthly cash sales and the accrual series derived from them.

EVERPIX_DIR=/path/to/Everpix-Intelligence npm run fixture:everpix
npm run demo:everpix

The dataset carries no explicit license grant, so it is read from your own checkout rather than vendored here; fixtures/everpix/generated/ is gitignored and the Everpix tests skip when it is absent.

recognize reproduces Everpix's published accrual series from cash alone, straight-line over a twelve-month term, to floating-point precision — that check passing is what validates the op. The rest of the suite is expected to fail: it encodes what a healthy subscription business looks like, and Everpix was not one. Recognized revenue never covered AWS in any of the thirteen measured months, storage never once fell, and the cash and accrual bases disagree by 188% at their worst.

npm install          # peer deps double as the test toolchain
npm run build        # tsc -> dist/
npm test             # builds, then runs the node:test suite
npm run docs         # regenerate docs/index.html

The documentation page is one self-contained file. Its prose lives in tools/docs/content.mjs; the API reference is read out of src/index.ts by the TypeScript compiler at build time, so it cannot describe an export the package no longer has — and an export with no JSDoc shows up on the page saying so. GitHub Actions rebuilds and publishes it on every push to master.

Requires Node ≥ 20.19. Tests run against the built dist/, so npm test builds first.

Extracted from the gestaltbi-core client at v0.1.0 — this repo is now the canonical home and gestaltbi-core consumes it as a remote dependency. Behavior preserved verbatim from the in-tree predecessor, plus a couple of latent fixes (Math.max([1,2]) → Math.max(1,2) in aggregate's finalize). API will likely settle as the rete.js editor and the dbt-driven structure.json generator land.

Since v0.1.0, additive: the recognize / cohort / assert ops, a real diffcalc (it was an empty stub), ratio aggregation, the check library, StructureDirectory, and a test suite. Two latent bugs fixed:

• aggregate's median took the middle of an unsorted, scan-order array.
• Processor.initializeAggregator let a malformed dimension hierarchy throw out of workOn, taking down graphs that never touch the cube. It now records cubeError and leaves liveCube() to raise.

Ops now resolve the time column through resolveTimeColumn, which accepts both tag vocabularies in circulation: config repos authored for gestaltbi-core (see GestaltBI/sample-config) rename the date column to the canonical code uatu:date via mapping.json and tag it uatu:timedimension, while @gestaltbi/infer emits uatu:dimension / uatu:dimension:time. enhance's cumsum previously read the literal 'uatu:date' key, which is correct for the former and silently a no-op for the latter; it now handles both, and falls back to the literal key when no structure is loaded. Reconciling the two vocabularies at the infer end is still open.

enhance also gained an opt-in nullSafe option. By default a missing operand still reads as the arithmetic identity, so revenue - null returns revenue — a missing cost looks like a healthy margin. nullSafe: true propagates null instead. The default is unchanged.

Known issue, not fixed here: Processor.process() never pushes to this.done, so the if (this.done.indexOf(req) < 0) guard is dead and required processes are re-applied on every path that reaches them. Linear chains are unaffected; a diamond-shaped graph will apply the shared ancestor twice, and since ops mutate rows in place that double-counts. The one-line fix changes evaluation semantics for existing gestaltbi-core graphs, so it wants a decision rather than a drive-by patch.

MIT