# Streams im Objektmodus: Datenpipelines jenseits von Bytes

URL: https://www.mikebild.dev/de/blog/nodejs-streams-im-objektmodus/

Wer einen Stream auf `objectMode: true` umstellt und dabei sein gewohntes `highWaterMark: 16384` mitnimmt, puffert keine 16 KiB mehr, sondern 16384 Datensätze. Gleiche Zahl, andere Einheit, ein Speicherbedarf um Größenordnungen daneben. Der Objektmodus sieht aus wie eine beiläufige Konstruktoroption, aber er ändert, was ein Stream überhaupt durch die Pipeline schiebt – und damit auch, wann gedrosselt wird.

Im [früheren Beitrag zu Node.js-Streams und Backpressure](https://www.mikebild.dev/de/blog/nodejs-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. Sobald aber keine Dateien kopiert, sondern Datensätze verarbeitet werden – CSV-Zeilen, JSON-Objekte, Log-Einträge –, reicht dieses Bild nicht mehr weit genug.

Arbeitet ein Stream im Objektmodus, 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](https://www.mikebild.dev/de/blog/nodejs-training-event-loop-architektur/) 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:

```javascript
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 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: `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 `push`en 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:

```javascript
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.

```javascript
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:

```javascript
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.

```javascript
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:

```mermaid
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.

```javascript
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

`highWaterMark` bedeutet im Objektmodus etwas anderes als im Bytemodus – das wird regelmäßig überlesen, mit spürbaren Folgen. `highWaterMark` ist die Schwelle, ab der ein Stream als „voll genug“ gilt und Backpressure auslöst. Im Bytemodus wird diese Schwelle in Bytes gemessen; 2022 waren allgemein 16 KiB der Default, seit Node 22 sind es 64 KiB. Im Objektmodus wird sie in Objekten gemessen, weiterhin 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. Der allgemeine Default hängt von der Node-Version ab und liegt seit Node 22 bei 64 KiB.
- 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.

```javascript
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:

```javascript
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

Der erste Fallstrick ist das Mischen von Byte- und Objektmodus in einer Pipeline. 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.

Der zweite betrifft `null` als legitimen 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](https://www.mikebild.dev/de/blog/nodejs-buffers-und-binaerdaten/) verstanden hat und den Objektmodus dazunimmt, hat beide Hälften der Stream-Welt beisammen – die rohe und die strukturierte.

## Weiterführende Quellen

- [Node.js Stream API – Object mode](https://nodejs.org/api/stream.html#object-mode)
- [Node.js Stream API – Buffering](https://nodejs.org/api/stream.html#buffering)
- [Node.js Stream API – stream.Readable.from](https://nodejs.org/api/stream.html#streamreadablefromiterable-options)
- [Node.js Stream API – readable[Symbol.asyncIterator]](https://nodejs.org/api/stream.html#readablesymbolasynciterator)
- [Node.js Stream API – Class: stream.Transform](https://nodejs.org/api/stream.html#class-streamtransform)