Skip to main content

Transformations

The simplest shape: events in, events out. One event per source event, same temporal key, sometimes a different schema. These are the operators that preserve the event grid — they don't bucket, window, or aggregate.

For operations that change the event grid (rolling windows, fixed-bucket aggregation, alignment), see Concepts → Windowing and the Aggregation, Alignment, and Rolling Windows operator pages.

Two flavors:

  • Per-event-only — operate on one event at a time, no memory. map, select, rename, collapse, filter, plus set / merge on individual events.
  • Per-event-with-tiny-state — look at the previous event (or a small accumulator) to compute each output. diff, rate, pctChange, cumulative, scan, shift.

Coming from pandas: this section covers .assign(), .rename(), [cols], .where() (≈ filter), .diff(), .pct_change(), .cumsum(), .shift(). Coming from JavaScript arrays: map, filter do exactly what you'd expect; the rest are time-series-aware.

Per-event-only​

map(nextSchema, fn) — transform each event​

The most general. Pass the output schema (so the result type is narrow) and a function that takes a source event and returns the new one. The schema argument is mandatory even when the shape doesn't change — pond-ts always wants the next schema explicit.

// Same schema; just transform the cpu value.
const scaled = cpu.map(cpu.schema, (e) => {
  const value = e.get('cpu');
  return value === undefined ? e : e.set('cpu', value * 100);
});

// New schema — adds a "flagged" column.
const flaggedSchema = [
  ...cpu.schema,
  { name: 'flagged', kind: 'boolean' },
] as const;
const flagged = cpu.map(flaggedSchema, (e) =>
  e.merge({ flagged: (e.get('cpu') ?? 0) > 0.9 }),
);

Event instances are immutable; set returns a new Event with the field updated. merge(patch) works the same way for multi-field updates and can introduce new columns.

filter(predicate) — drop events​

Predicate gets an event and an index, returns true to keep:

const active = cpu.filter((e) => e.get('cpu') !== undefined);
const apiOnly = events.filter((e) => e.get('host')?.startsWith('api'));

Same schema in, same schema out; just fewer rows. See Cleaning data for filter as a "remove bad rows" pattern.

filter also takes an expression in place of the function, such as cpu.filter(col('cpu').gt(0.5)). It runs over whole columns and is usually several times faster.

select(...keys) — keep only some columns​

Schema-narrowing — type-level too:

const just = full.select('cpu', 'host');
// just.schema === [time, cpu, host]; the other columns are gone.
just.at(0)?.get('cpu'); // still narrows to number | undefined

Pass the names as separate arguments. A name that isn't a payload column throws a RangeError listing the valid columns, and an array argument (select(['cpu', 'host'])) throws a TypeError — spread it: select(...names). Output columns keep schema order; a repeated name selects its column once.

rename(mapping) — rename payload columns​

const renamed = cpu.rename({ cpu: 'usage' });
renamed.at(0)?.get('usage'); // number | undefined
// renamed.schema has 'usage' where 'cpu' was.

The temporal key column can't be renamed.

collapse(keys, output, reducer) — combine columns into one​

Reducer receives a typed record of just the named columns, returns a single value:

// Average of cpu and requests into a "load" column.
const summary = metrics.collapse(
  ['cpu', 'requests'],
  'load',
  ({ cpu, requests }) => ((cpu ?? 0) + (requests ?? 0)) / 2,
);
// schema: [time, ...other source columns, load]

The named columns are removed by default; { append: true } keeps them and adds the new one alongside.

Event.set(field, value) and Event.merge(patch)​

Work directly on an Event. You'll use them inside map callbacks:

events.map((e) => e.set('cpuPct', (e.get('cpu') ?? 0) * 100));

events.map((e) =>
  e.merge({
    cpu: (e.get('cpu') ?? 0) * 100,
    flagged: (e.get('cpu') ?? 0) > 0.9,
  }),
);

set returns the same key; merge does too. Both are immutable.

pond-rs has no Event objects to set or merge: a Row is a read-only view. Use map_numbers, with_column or collapse instead.

Per-event-with-tiny-state​

These look at one event plus the previous (or an accumulator) to compute each output. Length is preserved; the first event has no predecessor, so target columns are undefined (or you can drop the first event with { drop: true }).

diff(cols) — per-event delta​

const deltas = series.diff('requests');
// requests = curr - prev. First event: undefined.
const both = series.diff(['cpu', 'mem']);
const dropped = series.diff('requests', { drop: true });

diff, rate and pctChange replace each named column with its change. To keep the source and add the change beside it, name an output (one source column only; the name must be new):

const withChange = series.diff('requests', { output: 'requestsDelta' });
// requests unchanged; requestsDelta = curr - prev (optional number)

rate(cols) — delta per second​

diff divided by the time gap (in seconds). When two consecutive events share a timestamp (dt = 0), rate returns undefined rather than dividing by zero — filter per-producer first if you have concurrent emissions.

const rps = series.rate('requests');

pctChange(cols) — relative change​

(curr - prev) / prev. Time-gap-independent. Returns undefined when the previous value is 0.

const rets = prices.pctChange('price');

cumulative(spec) — running total / max / min / count / rises / falls​

const running = series.cumulative({ requests: 'sum', cpu: 'max' });
const errors = series.cumulative({ errors: 'count' });
const product = series.cumulative({ factor: (acc, v) => acc * v });

When a source value is undefined, the accumulator carries forward its previous value — no NaN propagation.

Rises and falls: 'increase' / 'decrease'​

'increase' is the running total of rises between values, 'decrease' the running total of falls (a positive number). Give a threshold to ignore small wiggles: a change counts only once the value has moved at least that far from the last value that counted, in either direction, so slow drift still adds up. That is elevation gain with a dead band, or the use of a noisy meter or level.

// ele: 100, 101, 104, (gap), 102, 99, 103, 108
const gain = track.cumulative({ ele: { using: 'increase', threshold: 3 } });
// → 0, 0, 4, 4, 4, 4, 8, 13
const loss = track.cumulative({ ele: { using: 'decrease', threshold: 3 } });
// → 0, 0, 0, 0, 0, 5, 5, 5
const everyRise = track.cumulative({ ele: 'increase' });

The first value starts the total at 0, and both directions share one reference point. Because it is a named accumulator rather than a step function, @pond-ts/rs runs it without calling back into JavaScript.

scan(source, step, init, options?) — typed-accumulator running fold​

scan is the general form of cumulative — the classic mapAccumL. cumulative locks three things together (the accumulator is the output is a number, written in place); scan decouples them: the accumulator A can be any value, the numeric output is separate, and you choose whether to replace the source or append a new column.

// running sum — the cumulative special case, replacing in place:
series.scan('work', (acc, v) => [acc + v, acc + v], 0);
//   ≡ series.cumulative({ work: 'sum' })

// typed accumulator into a NEW column: hysteresis elevation gain
// carries (ref, gain) but emits only the gain. (This exact rule is built
// in as cumulative({ ele: { using: 'increase', threshold: 3 } }); the scan
// form shows how to write your own.)
const T = 3; // ±3 m deadband
const withGain = track.scan<'cumGain', { ref: number | null; gain: number }>(
  'ele',
  (acc, ele) => {
    if (acc.ref === null) return [{ ref: ele, gain: 0 }, 0];
    const d = ele - acc.ref;
    if (d >= T) return [{ ref: ele, gain: acc.gain + d }, acc.gain + d];
    if (d <= -T) return [{ ref: ele, gain: acc.gain }, acc.gain];
    return [acc, acc.gain]; // within deadband — carry
  },
  { ref: null, gain: 0 },
  { output: 'cumGain' },
);

step(acc, value, i) returns [nextAcc, output]. The accumulator A is inferred from init. With no options.output the source column is replaced (widened to optional number, as cumulative does); with options.output a new column of that name is appended and the source is left intact (the name must not already exist).

Missing cells inherit cumulative's behavior exactly: a missing / undefined source cell does not call step — the accumulator is held and the row re-emits the last output (so it holds flat across a gap), undefined only until the first defined value. A stored NaN is a defined number and is passed to step. On a multi-entity series the accumulator interleaves across entities; scope it per entity with partitionBy, run scan inside each partition, and fan back in with collect:

// One running total per host, not a single interleaved accumulator.
const perHost = series
  .partitionBy('host')
  .scan('requests', (acc, v) => [acc + v, acc + v], 0)
  .collect();

split = scan + byColumn. The reason scan exists: it isolates the order-dependent state into a column so value-axis aggregation (byColumn) can stay pure and order-free. Materialize the carried state with scan, then segment statelessly — e.g. per-kilometre split gain is last − first of the scanned cumGain in each distance bin. See Value axis for the mental model behind this composition.

tail(duration) — keep the trailing window​

tail is technically a temporal selection rather than a per-event op, but it's tiny and lives next to these in practice — keep events whose begin() is greater than lastEvent.begin() - duration. Same schema, fewer rows.

const recent = series.tail('30s');
const all = series.tail(); // identity (whole series) when no duration

Often paired with reduce for "current state" readouts — see Rolling windows → tail for the composition.

shift(cols, n) — lag / lead​

Move column values forward (positive n = lag) or backward (negative n = lead). Vacated slots become undefined. Keys are unchanged.

const lagged = series.shift('value', 1);
const lead = series.shift('value', -1);
const multi = series.shift(['cpu', 'mem'], 2);

For time-based shifts on irregular data, align to a regular grid first (Alignment), then shift by the corresponding number of events.

offsetTime(by) — move the keys in time​

Moves every key by a constant — milliseconds or a duration string, negative for earlier. A duration is a fixed length: '1d' is 24 hours, not a calendar day, so across a daylight-saving change an overlay is an hour off. Values stay on their rows; a timeRange / interval key moves both edges and keeps its labels. Value columns are shared, not copied, so this costs one pass over the key.

// Overlay yesterday on today: yesterday's rows, keyed a day later.
const yesterdayOnToday = yesterday.offsetTime('1d');
const skewFixed = sensor.offsetTime(-250); // clock ran 250 ms fast

shift moves values by rows and leaves keys alone; offsetTime moves keys by time and leaves values alone. If you only want a chart to draw a column somewhere else — a 1-minute bar's close at the bar's end — use the layer's xOffset prop in @pond-ts/charts instead: the series stays as it is.

Method comparison​

MethodOutput
map(nextSchema, fn)New schema possible; one event in, one out
filter(pred)Same schema; subset of rows
select(...keys)Schema narrowed to chosen columns
rename(mapping)Schema with renamed columns
collapse(cols, ...)Schema replaces several columns with one
diff(cols)Same schema; per-event delta column
rate(cols)Same schema; delta-per-second column
pctChange(cols)Same schema; relative-change column
cumulative(spec)Same schema; running aggregate column
scan(src, …)Replace source, or append output column
tail(duration)Same schema; trailing temporal slice
shift(cols, n)Same schema; column values shifted
offsetTime(by)Same schema; every key moved by by
Event.set(field, v)Single field on one event
Event.merge(patch)Multi-field patch on one event