Streams im Objektmodus: Datenpipelines jenseits von Bytes

In meinen Node.js-Schulungen kommt fast immer derselbe Moment. Wir haben über Streams gesprochen, über Chunks, über Backpressure – und irgendjemand fragt: „Und was mache ich, wenn ich gar keine Dateien kopiere, sondern Datensätze verarbeite? CSV-Zeilen, JSON-Objekte, Log-Einträge?" Genau das ist die Frage, um die es hier geht. Im früheren Beitrag zu Node.js-Streams und Backpressure drehte sich alles um Bytes: Buffer, die durch eine Pipe fließen, und die Kunst, den Fluss zu drosseln, bevor der Speicher überläuft. Das ist der Standardfall für Datei- und Netzwerk-I/O. Aber Streams können mehr, und dieses Mehr heißt objectMode.

Sobald ein Stream im Objektmodus arbeitet, transportiert er keine Buffer und keine Strings mehr, sondern beliebige JavaScript-Werte – ganze Objekte, Arrays, Zahlen. Damit wird aus dem Werkzeug für rohe Byteströme ein Werkzeug für Datenverarbeitung. Und das Schöne daran: Die Mechanik bleibt dieselbe. Man verkettet Stufen, jede Stufe macht eine Sache, und am Ende steht eine Pipeline, die einen Datensatz nach dem anderen durchreicht, ohne alles auf einmal in den Speicher zu laden.

Was der Objektmodus wirklich ändert

Ein Stream im Standardmodus kennt genau zwei Datentypen: Buffer und String. Alles, was man hineinschiebt, muss eines von beiden sein. Der Objektmodus hebt diese Einschränkung auf. Ein new stream.Transform({ objectMode: true }) akzeptiert und liefert beliebige JS-Werte. Die einzige Ausnahme ist null – dieser Wert ist reserviert und signalisiert das Ende des Streams. Dazu später mehr, denn genau hier lauert ein Fallstrick.

Der Objektmodus wird pro Stream über die Konstruktoroption gesetzt, und der Standardwert ist false. Man aktiviert ihn also bewusst. In den Beispielen aus meinem Node.js-Repository taucht das Muster bereits durchgehend auf – dort allerdings noch in der ES5-Schreibweise mit util.inherits und Readable.call(this, { objectMode: true }). Ein Readable, der zwei kleine Objekte durchreicht, sieht dort so aus:

const { Readable } = require('stream');
const util = require('util');

// ES5 pattern from the training repo
const ReadStream = function () {
  Readable.call(this, { objectMode: true });
  this.data = [
    { id: 0, value: 1 },
    { id: 1, value: 2 },
  ];
  this.curIndex = 0;
};
util.inherits(ReadStream, Readable);

ReadStream.prototype._read = function () {
  if (this.curIndex === this.data.length) {
    return this.push(null); // signals end of stream
  }
  this.push(this.data[this.curIndex++]);
};

Das funktioniert und ist historisch gewachsen. Ich zeige es im Unterricht trotzdem lieber in der modernen Klassensyntax, weil sie näher an dem liegt, was man heute in Codebasen antrifft. Mit class ... extends Transform wird aus dem Konstruktor-Ritual ein super({ objectMode: true }), und die Verarbeitungslogik steht in einer klar benannten Methode.

Die eine Kernmethode: `_transform`

Ein Transform-Stream hat im Kern eine einzige Methode, die man selbst schreibt: _transform(chunk, encoding, callback). Für jeden eintreffenden Wert wird sie einmal aufgerufen. Man gibt Ergebnisse mit this.push(value) aus und meldet mit callback(), dass dieser Wert fertig verarbeitet ist. Im Objektmodus ist chunk der rohe JS-Wert – das Argument encoding wird ignoriert, weil es hier nichts zu dekodieren gibt.

Der entscheidende Punkt für Einsteiger: push und callback sind zwei getrennte Dinge. push schiebt Ausgaben in die nächste Stufe, callback signalisiert „ich bin mit diesem Eingabewert durch". Man kann pro Eingabe null-, ein- oder mehrmals pushen und ruft am Ende genau einmal callback auf. Ruft man nur callback() ohne push, verschwindet der Wert – das ist die natürliche Art, in einem Filter etwas wegzulassen.

Hier eine Verarbeitungsstufe, die rohe CSV-Zeilen zu Objekten parst:

const { Transform } = require('stream');

class ParseRow extends Transform {
  constructor() {
    super({ objectMode: true });
  }

  _transform(line, _encoding, callback) {
    const [id, name, status] = String(line).split(',');
    this.push({ id: Number(id), name, status });
    callback();
  }
}

Und eine Filterstufe, die nur aktive Datensätze durchlässt. Hier zeigt sich das Weglassen: Ist die Bedingung nicht erfüllt, wird nichts gepusht, sondern nur callback() aufgerufen.

const { Transform } = require('stream');

class FilterActive extends Transform {
  constructor() {
    super({ objectMode: true });
  }

  _transform(record, _encoding, callback) {
    if (record.status === 'active') {
      this.push(record);
    }
    // no push means the record is dropped
    callback();
  }
}

Eine Formatierungsstufe rundet das Bild ab. Sie nimmt das Objekt und formt eine Ausgabezeile daraus:

const { Transform } = require('stream');

class FormatOutput extends Transform {
  constructor() {
    super({ objectMode: true });
  }

  _transform(record, _encoding, callback) {
    this.push(`#${record.id} ${record.name}\n`);
    callback();
  }
}

Drei kleine Stufen, jede mit einer einzigen Verantwortung. Genau so entsteht Komponierbarkeit.

Stufen verketten: parsen, filtern, formatieren

Die eigentliche Pipeline ist dann fast unspektakulär – und das ist ein gutes Zeichen. Ich verkette die Stufen mit pipeline() aus dem stream-Modul, nicht mit einer Kette aus pipe(). Der Grund: pipeline() propagiert Fehler und räumt die beteiligten Streams auf, wenn einer scheitert. Eine reine pipe()-Kette tut das nicht – dort muss man Fehler und Cleanup an jeder Naht selbst behandeln, und genau das wird in der Praxis vergessen.

const { Readable, pipeline } = require('stream');

const csvLines = ['0,Ada,active', '1,Linus,inactive', '2,Grace,active'];

pipeline(
  Readable.from(csvLines),
  new ParseRow(),
  new FilterActive(),
  new FormatOutput(),
  process.stdout,
  (err) => {
    if (err) {
      console.error('pipeline failed:', err);
    } else {
      console.log('pipeline finished');
    }
  }
);

Die Quelle ist hier Readable.from(csvLines) – dazu gleich mehr. Die Senke ist process.stdout, ein ganz normaler Writable-Stream. Dazwischen fließt ein Datensatz nach dem anderen: erst als String, nach ParseRow als Objekt, durch FilterActive gefiltert, in FormatOutput wieder zu einer Zeile geformt. Zu keinem Zeitpunkt liegt die gesamte Datenmenge im Speicher – es ist immer nur ein kleiner Vorrat gepufferter Werte unterwegs.

Als Diagramm sieht dieser Datenfluss so aus:

flowchart LR
  A["Source<br/>Readable.from"] -->|"CSV line (string)"| B["Parse<br/>_transform"]
  B -->|"JS object"| C["Filter<br/>_transform"]
  C -->|"JS object"| D["Format<br/>_transform"]
  D -->|"output line"| E["Sink<br/>Writable"]

An den Kanten steht, was tatsächlich fließt: erst eine Zeile als String, dann JS-Objekte durch die mittleren Stufen, am Ende wieder Text. Jede Stufe kümmert sich nur um ihre eigene Umwandlung und weiß nichts von den Nachbarn.

Aggregieren mit `_flush`

Filtern und Umformen arbeiten Datensatz für Datensatz. Für Aggregationen – Summen, Zählungen, Gruppierungen – braucht man einen anderen Rhythmus: sammeln, während die Werte eintreffen, und das Ergebnis erst ausgeben, wenn die Quelle erschöpft ist. Dafür gibt es _flush(callback). Diese Methode wird genau einmal aufgerufen, kurz bevor der Stream endet. In _transform sammelt man, in _flush gibt man aus.

const { Transform } = require('stream');

class CountByStatus extends Transform {
  constructor() {
    super({ objectMode: true });
    this.counts = {};
  }

  _transform(record, _encoding, callback) {
    this.counts[record.status] = (this.counts[record.status] || 0) + 1;
    callback(); // nothing pushed yet, just accumulate
  }

  _flush(callback) {
    this.push(this.counts); // emit the aggregate once at the end
    callback();
  }
}

Diese Stufe schluckt viele Objekte und gibt am Ende ein einziges aus – das Aggregat. Sie zeigt gut, dass eine Stream-Stufe ihre Ein- und Ausgabemenge nicht eins zu eins halten muss.

Backpressure zählt jetzt Objekte

Ein Punkt sorgt in fast jeder Schulung für Verwirrung, deshalb sage ich es deutlich: highWaterMark bedeutet im Objektmodus etwas anderes als im Bytemodus. highWaterMark ist die Schwelle, ab der ein Stream als „voll genug" gilt und Backpressure auslöst. Im Bytemodus wird diese Schwelle in Bytes gemessen, mit einem Standardwert von 16384 Bytes, also 16 KiB. Im Objektmodus wird sie in Objekten gemessen, mit einem Standardwert von 16 Objekten.

Das ist keine Kleinigkeit. Wer aus dem Bytemodus kommt und highWaterMark: 16 liest, denkt reflexartig an 16 Bytes und hält das für absurd klein. Im Objektmodus sind es aber 16 gepufferte Objekte – ein völlig sinnvoller Vorrat. Umgekehrt kann ein aus Byte-Gewohnheit gesetztes highWaterMark: 16384 im Objektmodus bedeuten, dass man 16384 Objekte puffert, bevor Backpressure greift. Das ist selten gewollt.

Wichtig ist auch: highWaterMark ist eine Schwelle, kein harter Riegel. Der Puffer darf sie überschreiten – die Schwelle ist das Signal, ab dem push() den Wert false zurückgibt und nachgelagerte Stufen zum Drosseln auffordert. Bei pipeline() und pipe() geschieht dieses Pausieren und Fortsetzen automatisch. Man verlässt sich also im Normalfall auf dieses automatische Zusammenspiel und muss den Rückgabewert von push nur dann selbst auswerten, wenn man eine Quelle von Hand steuert. Backpressure funktioniert im Objektmodus genauso zuverlässig wie im Bytemodus, nur ist die Einheit eine andere.

Zur Abgrenzung die beiden Modi nebeneinander:

Der Bytemodus (Standard) transportiert Buffer oder String, highWaterMark zählt hier Bytes mit Standardwert 16 KiB, und die typischen Einsätze sind Datei-I/O, Netzwerk und Kompression – überall dort, wo rohe Byteströme durchlaufen.
Der Objektmodus (objectMode: true) transportiert beliebige JS-Werte außer null, highWaterMark zählt hier Objekte mit Standardwert 16, und die typischen Einsätze sind Datenpipelines aus parsen, filtern, transformieren und aggregieren.

Die Brücke zu Async-Iteratoren

Bis hierhin habe ich Streams als Klassen gebaut. Das ist nicht immer nötig, und oft ist es sogar umständlich. Node bietet zwei moderne Verbindungsstücke, die den Objektmodus mit der Welt der Async-Iteratoren zusammenbringen: Readable.from(iterable) und for await ... of.

stream.Readable.from(iterable) nimmt ein beliebiges Iterable oder Async-Iterable – ein Array, einen Generator, einen Async-Generator – und macht daraus einen Readable-Stream. Standardmäßig arbeitet dieser Stream im Objektmodus. Das ist die eleganteste Art, eine Datenquelle in eine Pipeline einzuspeisen, ohne eine eigene Readable-Klasse zu schreiben. Genau das habe ich oben mit Readable.from(csvLines) bereits genutzt.

Die Gegenrichtung ist mindestens so praktisch: Ein Readable-Stream ist selbst ein Async-Iterable. Man kann ihn also direkt mit for await ... of konsumieren, und das Sprachkonstrukt kümmert sich um Pausieren, Fortsetzen, Backpressure und das Ende – alles automatisch.

const { Readable } = require('stream');

async function* generateRecords() {
  yield { id: 0, value: 1 };
  yield { id: 1, value: 2 };
  yield { id: 2, value: 3 };
}

async function main() {
  const source = Readable.from(generateRecords());

  for await (const record of source) {
    console.log('received:', record);
  }
}

main();

Man kann Async-Generatoren sogar als Pipeline-Stufe einsetzen. Ein Generator, der eine Quelle konsumiert und gefilterte Werte weiterreicht, ersetzt eine ganze Transform-Klasse durch wenige Zeilen:

async function* filterEven(source) {
  for await (const n of source) {
    if (n % 2 === 0) {
      yield n;
    }
  }
}

Ob man eine Transform-Klasse schreibt oder einen Async-Generator, ist zu einem guten Teil Geschmackssache. Ich greife zur Klasse, wenn ich internen Zustand über viele Werte halten muss, etwa bei einer Aggregation mit _flush. Für reine Filter und Umformungen sind Async-Generatoren oft der kürzere, besser lesbare Weg.

Zwei Fallstricke, die immer wieder zuschlagen

Byte- und Objektmodus in einer Pipeline mischen. Der Klassiker: Man beginnt mit fs.createReadStream(...), das ist ein Byte-Stream und liefert Buffer. Direkt danach hängt man eine Objektmodus-Stufe, die mit record.status arbeiten will – und bekommt einen Buffer statt eines Objekts. Zwischen einem Byte-Readable und der ersten Objektstufe muss immer eine Parse-Schicht liegen, die Buffer zu Werten dekodiert. Die umgekehrte Naht ist genauso heikel: Schiebt man ein Objekt in einen Nicht-Objektmodus-Writable, quittiert Node das mit einem Typfehler (ERR_INVALID_ARG_TYPE), der einen String, Buffer oder ein Uint8Array verlangt. Man sollte sich an jeder Nahtstelle bewusst machen, was für ein Datentyp gerade fließt.

null als legitimer Nutzwert. Weil push(null) das Ende des Streams bedeutet, kann man einen echten null-Wert nicht durch einen Objektmodus-Stream schicken – er würde als Schlusssignal missverstanden und die restlichen Daten abschneiden. Wenn null in den Nutzdaten eine echte Bedeutung hat, braucht es ein Ersatzobjekt, einen sogenannten Sentinel, etwa { value: null } statt eines nackten null.

Ein dritter Punkt gehört zur Sorgfalt: Fehler in _transform gibt man an callback(err) weiter, statt sie zu werfen. Nur so landen sie im Fehlerkanal des Streams und werden von pipeline() sauber propagiert. Ein geworfener Fehler mitten in der asynchronen Verarbeitung ist deutlich schwerer einzufangen – auch das ist ein Argument für pipeline() gegenüber einer nackten pipe()-Kette.

Fazit

Der Objektmodus verwandelt Node.js-Streams von einem I/O-Werkzeug für Bytes in ein Werkzeug für Datenverarbeitung. Die Mechanik bleibt vertraut – _transform mit push und callback, _flush fürs Aggregieren, Verkettung über pipeline() –, aber der Inhalt sind jetzt ganze JS-Werte. Damit lassen sich Pipelines aus parsen, filtern, transformieren und aggregieren als getrennte, komponierbare Stufen bauen, die immer nur einen kleinen Vorrat an Datensätzen im Speicher halten.

Zwei Dinge nehme ich als Kernbotschaft mit. Erstens: highWaterMark zählt im Objektmodus Objekte, nicht Bytes – wer das verwechselt, dimensioniert seinen Puffer um Größenordnungen falsch. Zweitens: Readable.from und for await ... of sind die moderne Brücke zwischen Streams und Async-Iteratoren und ersparen einem in vielen Fällen die eigene Stream-Klasse. Wer den byteorientierten Teil aus dem Beitrag zu Buffers und Binärdaten verstanden hat und den Objektmodus dazunimmt, hat beide Hälften der Stream-Welt beisammen – die rohe und die strukturierte.