Skip to main content

Cleaning Data

Real ingestion is messy. Timestamps drift, values arrive null, duplicate rows slip in, the source produces gaps that aggregation later turns into undefined cells. This page is the place to look when you need to fix a TimeSeries before downstream transforms can reason over it.

What's covered:

Schema first — required: false​

Most cleaning starts with knowing which columns can have missing cells. Declare those columns required: false in your schema — otherwise the type system narrows their cell values to T (not T | undefined), and fill/dedupe won't have anything to operate on:

const schema = [
  { name: 'time', kind: 'time' },
  { name: 'cpu', kind: 'number', required: false }, // ← optional
  { name: 'host', kind: 'string' }, // ← required (always present)
] as const;

When ingesting JSON or CSV-derived rows, missing cells arrive as null on the wire and become undefined after TimeSeries.fromJSON(...). (For row-array construction with new TimeSeries({ rows }), the RowForSchema type currently doesn't honor required: false — go through fromJSON with null cells when you need optional values, or see the known limitation in the 0.8.2 changelog.)

Filter out bad rows​

filter() is the simplest cleaning tool — keep events that match a predicate, drop the rest. Same schema in, same schema out, fewer rows.

// Drop events whose value is missing.
const present = cpu.filter((e) => e.get('cpu') !== undefined);

// Drop obvious sensor errors (negative CPU, value over 100%).
const sane = cpu.filter((e) => {
  const v = e.get('cpu');
  return v !== undefined && v >= 0 && v <= 1;
});

// Drop events from a known-bad host.
const trusted = events.filter((e) => e.get('host') !== 'staging-1');

Filtering early means downstream transforms never see the bad rows. For exploration, also pull just the bad rows into a separate series to inspect:

const bad = cpu.filter((e) => {
  const v = e.get('cpu');
  return v === undefined || v < 0 || v > 1;
});
console.log(bad.length, 'suspect rows');

Fill the gaps​

fill() replaces undefined values that aggregation, alignment, or upstream sources leave behind. The typical post-aggregate cleanup.

Strategies​

  • 'hold' — forward-fill: carry the last known value. Leading undefined runs stay unfilled (nothing to carry).
  • 'bfill' — backward-fill: pull in the next known value. Trailing undefined runs stay unfilled.
  • 'linear' — time-interpolate between the two nearest known values on either side. Leading and trailing undefined runs stay unfilled (no extrapolation).
  • 'zero' — fill with 0. Numeric columns only.

A single strategy applies to every value column:

series.fill('hold');
series.fill('linear');
series.fill('zero');

Per-column mapping targets specific columns; other columns pass through with their undefineds intact:

series.fill({ cpu: 'linear', host: 'hold' });

Literal fill values (anything that's not a strategy string) set the column to that value:

series.fill({ cpu: 0, host: 'unknown' });

Gap caps — limit (cells) and maxGap (duration)​

fill() operates on gaps — runs of consecutive undefined cells in one column. Both options cap how big a gap can be before fill gives up:

// Count-based: gap of at most 3 cells fits the cap
series.fill('hold', { limit: 3 });

// Duration-based: gap whose temporal span (prev-known to next-known)
// is at most 5 minutes fits the cap
series.fill('linear', { maxGap: '5m' });

// Both — the more restrictive wins
series.fill('linear', { limit: 3, maxGap: '5m' });

All-or-nothing semantics. A gap either fits the caps or it doesn't:

  • Gap fits → fill the entire gap with the strategy.
  • Gap exceeds either cap → leave the entire gap unfilled.

This is the v0.9.0 default. Earlier versions partially filled (limit: 3 on a 5-cell gap filled 3, left 2 unfilled). Partial fill propagated stale 'hold' values past their useful lifetime and produced misleading 'linear' interpolations across long outages. All-or-nothing is honest: if the gap is too big to bridge, leave the unknown unknown.

// Series with one 10-minute outage gap:
const cpu = series.fill('linear', { maxGap: '5m' });
// 10-min gap exceeds 5-min cap → outage remains undefined.
// A naive linear interp would have invented a smooth slope across
// data we don't have.

Typical post-aggregate pipeline​

const hourly = series
  .aggregate(Sequence.every('1h'), { cpu: 'avg', host: 'last' })
  .fill({ cpu: 'linear', host: 'hold' }, { maxGap: '3h' });

Multi-entity series — see partitionBy​

fill() walks one chronological event sequence. On a multi-entity series (events for multiple hosts interleaved by time), neighbors cross entity boundaries — host-A's missing cell would interpolate against host-B's value as a "neighbor."

Use partitionBy(col) to scope fill to within each entity:

series.partitionBy('host').fill({ cpu: 'linear' }).collect();

See Concepts → Partitioning for the full hazard discussion.

Deduplicate​

dedupe() collapses events that share a key. Real-world ingest produces duplicates: WebSocket replays, Kafka at-least-once, retried HTTP fetches, polling overlaps. The default resolution is last-wins (matches the "newer event supersedes" intuition).

Multi-entity hazard

For series carrying multiple entities (host, region, device id), bare dedupe() will collapse events from different entities at the same timestamp as if they were duplicates of each other. Always pair it with partitionBy on multi-entity series before reaching for the operator.

For single-entity series, the default key is the event's full key (timestamp for time-keyed series; begin+end for time-range and interval keys; the labeled value is part of identity for intervals):

// Single-entity series — collapse same-timestamp duplicates, keep last.
const clean = events.dedupe();

Multi-entity dedupe — partition first​

If your series carries multiple entities, you almost always want (time + entity) as the dedupe key, not time alone. Bare dedupe() would collapse host=A@t and host=B@t as if they were duplicates of each other.

Pair it with partitionBy(col):

// Per-host dedupe — same time AND same host is the duplicate key.
const clean = events.partitionBy('host').dedupe({ keep: 'last' }).collect();

Resolution policies — keep​

keepBehavior
'last' (default)Keep the last event at each timestamp. Matches WebSocket replay / "newer wins" intuition.
'first'Keep the first event encountered.
'error'Throw on the first duplicate seen. Useful when duplicates indicate an upstream shape problem.
'drop'Discard every event at any duplicate timestamp. Conservative — never invents a survivor.
{ min: 'col' }Keep the event with the smallest value at the named numeric column. Ties keep first.
{ max: 'col' }Keep the event with the largest value at the named numeric column.
(events) => EventCustom resolver. Receives all duplicates (length ≥ 2), returns one. Use for merge logic.
// Throw on duplicates — useful in ingestion pipelines that want loud failures.
events.dedupe({ keep: 'error' });

// Keep the highest-CPU sample when two reports race at the same instant.
events.dedupe({ keep: { max: 'cpu' } });

// Custom merge: average duplicate cpu samples; keep the last host name.
events.dedupe({
  keep: (events) => {
    const last = events[events.length - 1];
    const avg =
      events.reduce((a, e) => a + (e.get('cpu') ?? 0), 0) / events.length;
    return last.set('cpu', avg);
  },
});

Chains compose​

dedupe returns the same schema, so it sits naturally before any other transform:

events
  .partitionBy('host')
  .dedupe({ keep: 'last' })
  .fill({ cpu: 'linear' })
  .rolling('5m', { cpu: 'avg' })
  .collect();

Regularize to a grid — materialize​

When source events arrive at irregular times — polling drift, mixed sample rates, sparse alerts — you often want to project the data onto a regular grid (one row per minute, one row per second, etc.) before downstream interpolation, charting, or alignment with another series. materialize does exactly this: emit one time-keyed row per sequence bucket, populated from a chosen source event in the bucket, with undefined cells where no source event existed.

series.materialize(Sequence.every('1m'));
// → one row per minute. Empty buckets emit undefined for all
//   value columns. No fill policy applied — that's a separate step.

materialize is the missing piece between dedupe and fill:

OperatorCreates rows?Picks fill values?
dedupenono (keeps survivor)
materializeyesno
fillnoyes
alignyesyes (mandatory)

align smooshes "make the grid" with "choose how to fill it" — the fill method ('hold' or 'linear') is mandatory. materialize does only the first step, leaving fill policy as an explicit follow-up. Use align when you want both at once; use materialize when you want gap-capped fill afterwards.

Options​

  • sample ('begin' | 'center' | 'end', default 'begin') — bucket anchor for the output time. Mirrors align's convention.
  • select ('first' | 'last' | 'nearest', default 'last') — which source event in each bucket wins. 'first' / 'last' pick the boundary event by begin() order. 'nearest' picks the source event whose begin() is closest to the bucket's sample time among events in the bucket — it never reaches across into a neighboring bucket. All three modes use half-open [bucket.begin, bucket.end) membership; an empty bucket emits undefined regardless of select.
  • range (TemporalLike, default series.timeRange()) — bounded slice for procedural sequences (Sequence.every(...)).

Multi-entity — partition first​

On a series carrying multiple entities, the partitioned variant auto-populates the partition column on every output row, including rows from buckets where the partition had no source event:

events
  .partitionBy('host')
  .materialize(Sequence.every('1m'))
  .fill({ cpu: 'linear' }, { maxGap: '3m' })
  .collect();
// Every row carries its host even on empty buckets — no need for
// a follow-up `.fill({ host: 'hold' })` step that would fail when
// every event in a partition lies in a long-outage gap.

End-to-end multi-entity cleaning pipeline​

The canonical multi-host cleaning chain — covers the three most common ingest hazards (cross-entity neighbor leaks, duplicate rows from retried polls, fabrication across long outages) in one pipeline:

const schema = [
  { name: 'time', kind: 'time' },
  { name: 'cpu', kind: 'number', required: false },
  { name: 'host', kind: 'string' },
] as const;

const raw = TimeSeries.fromJSON({
  name: 'cpu',
  schema,
  rows /* JSON rows from your ingest */,
});

const cleaned = raw
  // Scope every following stateful op to within each host.
  .partitionBy('host')
  // Collapse duplicates that share (time, host); keep newest.
  .dedupe({ keep: 'last' })
  // Linear-interpolate cpu, but only across short gaps. A 30-minute
  // outage stays unfilled — no fabricated slope across data we don't
  // have.
  .fill({ cpu: 'linear' }, { maxGap: '5m' })
  // Materialize back to a regular TimeSeries.
  .collect();

Each step removes one class of error:

StepHazard removed
partitionBy('host')Stateful ops below stop crossing entity boundaries — host-A no longer interpolates against host-B's neighboring value.
dedupe({ keep: 'last' })Retried HTTP polls / WebSocket replays at the same (time, host) collapse to the newest.
fill({ cpu: 'linear' }, { maxGap: '5m' })Short gaps fill cleanly; long outages stay undefined — no invented slopes across data you don't have.
.collect()Returns a TimeSeries<S> ready for downstream rendering or aggregation.

Ingest edges​

Most "the data is wrong" problems happen at the boundary, not inside pipelines. The relevant docs:

Cleaning the data after ingest can't recover from a wrong-zone parse. Get the boundary right first.