Streams in Node.js: Große Dateien effizient verarbeiten

1. Übersicht

Das Laden ganzer Dateien mit fs.readFile in den RAM ist praktisch für kleine Dateien. Bei mehreren Gigabyte großen Log-Dateien oder CSV-Dateien mit Millionen von Zeilen führt dieser Ansatz jedoch zu RAM-Überlauf oder zu einem Prozessabsturz. Dieser Artikel zeigt, wie du große Dateien mit Streams in Node.js verarbeitest: Daten in Chunks lesen, Zeilen trennen, Daten filtern, komprimieren und in neue Dateien schreiben. Du erfährst auch, wie Backpressure funktioniert und wie du Fehler korrekt mit pipeline behandelst.

Alle Beispiele verwenden Node.js 20+ und ES-Module. Deine package.json muss "type": "module" enthalten. Du benötigst keine externen Bibliotheken, da fs, stream, zlib und readline bereits integriert sind.

2. Das Problem mit readFile

Zuerst erstellen wir eine Test-Datei orders.csv mit etwa 5 Millionen Zeilen:

// generate-orders.mjs
import { createWriteStream } from 'node:fs';
import { once } from 'node:events';

const out = createWriteStream('orders.csv');
out.write('id,customer,amount,status\n');

for (let i = 1; i <= 5_000_000; i++) {
  const status = i % 10 === 0 ? 'FAILED' : 'PAID';
  const ok = out.write(`${i},customer-${i % 1000},${(i % 500) + 0.99},${status}\n`);
  if (!ok) await once(out, 'drain'); // warte darauf, dass der Schreibpuffer geleert wird
}
out.end();

Dieser Code nutzt bereits Backpressure. Abschnitt 5 erklärt diesen Mechanismus im Detail. Die erzeugte Datei ist etwa 180 MB groß.

Jetzt zählen wir die FAILED-Bestellungen auf die “naive” Weise:

// naive-count.mjs
import { readFile } from 'node:fs/promises';

const content = await readFile('orders.csv', 'utf8');
const failed = content.split('\n').filter(line => line.endsWith(',FAILED')).length;

console.log('FAILED:', failed);
console.log('RSS MB:', Math.round(process.memoryUsage().rss / 1024 / 1024));

FAILED: 500000
RSS MB: 1046

Eine 180 MB große Datei benötigt etwa 1 GB RAM. Der Prozess muss gleichzeitig drei Dinge im Speicher halten: den ursprünglichen Buffer, die UTF-8-Zeichenkette und ein Array mit 5 Millionen Substrings. Wenn die Datei größer als etwa 2 GB wird, wirft readFile einen ERR_FS_FILE_TOO_LARGE-Fehler und kann die Datei nicht lesen.

3. Die vier grundlegenden Stream-Typen

Ein Stream ist Datenfluss in Teilen (Chunks). Node.js hat vier Stream-Typen:

Typ	Aufgabe	Beispiel
Readable	Quelle der Daten	fs.createReadStream, process.stdin
Writable	Ziel der Daten	fs.createWriteStream, process.stdout
Duplex	Lesen und Schreiben, bidirektional unabhängig	TCP-Socket
Transform	Duplex mit Ausgabe abhängig von Eingabe	zlib.createGzip()

Jeder Stream hat einen internen Buffer. Die Größe dieses Buffers wird durch highWaterMark bestimmt. Standardmäßig nutzt fs.createReadStream 64 KiB. Streams im Object-Modus nutzen 16 Objekte.

4. Große Dateien Chunk für Chunk lesen

4.1. for await auf einem Readable nutzen

Der eleganteste Weg, einen Readable zu verarbeiten, ist Async-Iteration:

// read-chunks.mjs
import { createReadStream } from 'node:fs';

const stream = createReadStream('orders.csv', { highWaterMark: 64 * 1024 });

let bytes = 0;
let chunks = 0;
for await (const chunk of stream) {
  bytes += chunk.length;
  chunks++;
}

console.log({ chunks, mb: (bytes / 1024 / 1024).toFixed(1) });
console.log('RSS MB:', Math.round(process.memoryUsage().rss / 1024 / 1024));

{ chunks: 2853, mb: 178.3 }
RSS MB: 52

Jedes chunk ist ein Buffer mit maximal 64 KiB. Die Schleife fordert das nächste Chunk nur an, nachdem das aktuelle verarbeitet wurde. Daher hängt der RAM-Verbrauch praktisch nicht von der Dateigröße ab.

4.2. Zeilen lesen mit readline

Chunks werden nach Bytes aufgeteilt, nicht nach Zeilen. Um Zeile für Zeile zu verarbeiten, nutze readline:

// count-failed.mjs
import { createReadStream } from 'node:fs';
import readline from 'node:readline';

const rl = readline.createInterface({
  input: createReadStream('orders.csv'),
  crlfDelay: Infinity, // behandle \r\n als einzelnen Zeilenumbruch
});

let failed = 0;
for await (const line of rl) {
  if (line.endsWith(',FAILED')) failed++;
}
console.log('FAILED:', failed);

FAILED: 500000

Das Ergebnis ist gleich wie in Abschnitt 2, aber der RAM liegt nur bei 50–60 MB. readline setzt automatisch Zeilen zusammen, die über Chunks verteilt sind, daher musst du dir darum keine Gedanken machen.

5. Dateien schreiben und Backpressure

Backpressure tritt auf, wenn die Datenquelle schneller ist als das Schreibziel. Festplatte, Netzwerk oder Gzip können alle langsamer sein als die Lesegeschwindigkeit. Die Funktion writable.write() gibt false zurück, wenn der interne Buffer highWaterMark überschreitet. Das ist ein Signal: “Stopp, warten auf das drain-Event”.

Hier ein Beispiel, das dieses Signal ignoriert:

// bad-write.mjs — NICHT so tun
import { createWriteStream } from 'node:fs';

const out = createWriteStream('big.txt');
for (let i = 0; i < 10_000_000; i++) {
  out.write(`line ${i}\n`); // ignoriere den Rückgabewert
}
out.end();

Die Schleife läuft synchron, daher werden alle 10 Millionen Strings in den Buffer von out gepuffert, bevor die Festplatte sie schreiben kann. Der RAM-Verbrauch steigt auf mehrere Gigabyte an. Um das zu beheben, prüfst du den Rückgabewert von write() und rufst await once(out, 'drain') auf – genau wie in generate-orders.mjs aus Abschnitt 2.

Wenn du Streams mit pipeline verbindest, wird Backpressure automatisch verarbeitet. Das ist der empfohlene Ansatz in den meisten Fällen.

6. Streams mit pipeline verbinden

6.1. Datei kopieren und komprimieren

// gzip-orders.mjs
import { createReadStream, createWriteStream } from 'node:fs';
import { createGzip } from 'node:zlib';
import { pipeline } from 'node:stream/promises';

await pipeline(
  createReadStream('orders.csv'),
  createGzip(),
  createWriteStream('orders.csv.gz'),
);
console.log('Done');

Done

pipeline aus node:stream/promises gibt ein Promise zurück. Das Promise wird erfüllt, wenn der letzte Stream das Schreiben beendet hat. Wenn einer der Streams einen Fehler hat – etwa weil die Datei nicht existiert oder die Festplatte voll ist – wird das Promise abgelehnt. In diesem Fall werden alle Streams zerstört, sodass keine File Descriptor verloren gehen.

Hinweis: Vermeide a.pipe(b).pipe(c). pipe leitet Fehler nicht an andere Streams weiter. Wenn a einen Fehler hat, bleibt c offen. Nutze stattdessen pipeline.

6.2. Transform zum Trennen von Zeilen schreiben

Angenommen, du möchtest FAILED-Bestellungen filtern und dann komprimieren. Du brauchst einen Schritt zum Trennen von Zeilen, der in pipeline eingefügt werden kann. Schreiben wir einen eigenen Transform:

// split-lines.mjs
import { Transform } from 'node:stream';
import { StringDecoder } from 'node:string_decoder';

export function splitLines() {
  const decoder = new StringDecoder('utf8');
  let tail = '';

  return new Transform({
    readableObjectMode: true, // Ausgabe: eine Zeile pro Objekt (string), nicht Buffer

    transform(chunk, _encoding, callback) {
      const lines = (tail + decoder.write(chunk)).split('\n');
      tail = lines.pop(); // letzter Teil könnte eine unvollständige Zeile sein
      for (const line of lines) this.push(line);
      callback();
    },

    flush(callback) {
      const rest = tail + decoder.end();
      if (rest) this.push(rest);
      callback();
    },
  });
}

Erklärung zu den einzelnen Teilen:

• readableObjectMode: true: Die Lesseite gibt jede Zeile als separates Objekt aus. Die Schreibseite empfängt weiterhin Buffer.
• StringDecoder: Ein Chunk könnte mitten durch ein mehrbyte UTF-8-Zeichen geschnitten werden, z. B. den Buchstaben “ệ” in “Nguyễn Văn Hiệp”. Wenn du chunk.toString() direkt aufrufst, entsteht das fehlerhafte Zeichen �. decoder.write() behält unvollständige Bytes und kombiniert sie beim nächsten Aufruf.
• tail: speichert den unvollständigen Zeilenteil zum Verketten mit dem nächsten Chunk.
• flush: läuft, wenn die Eingabe endet. Dieser Schritt gibt die letzte Zeile aus, falls die Datei nicht mit \n endet.

6.3. Filtern mit Async-Generator

pipeline akzeptiert einen Async-Generator als Schritt. Dadurch kannst du filtern, ohne eine separate Transform-Klasse schreiben zu müssen:

// export-failed.mjs
import { createReadStream, createWriteStream } from 'node:fs';
import { createGzip } from 'node:zlib';
import { pipeline } from 'node:stream/promises';
import { splitLines } from './split-lines.mjs';

let count = 0;

await pipeline(
  createReadStream('orders.csv'),
  splitLines(),
  async function* (lines) {
    for await (const line of lines) {
      if (line.endsWith(',FAILED')) {
        count++;
        yield line + '\n';
      }
    }
  },
  createGzip(),
  createWriteStream('failed-orders.csv.gz'),
);

console.log(`${count} fehlerhafte Bestellungen exportiert`);

500000 fehlerhafte Bestellungen exportiert

Daten fließen durch fünf Schritte. Jeder Schritt behält nur wenige Chunks im Speicher. Wenn Gzip langsam ist, wartet der Generator bei yield und splitLines empfängt keine neuen Chunks. Das bedeutet, dass createReadStream das Lesen von der Festplatte vorübergehend einstellt. Das ist Backpressure – automatisch!

6.4. Mit AbortSignal zwischendurch abbrechen

Bei längeren Aufgaben solltest du eine Abbruch- oder Timeout-Möglichkeit anbieten:

// export-with-timeout.mjs
import { createReadStream, createWriteStream } from 'node:fs';
import { createGzip } from 'node:zlib';
import { pipeline } from 'node:stream/promises';

try {
  await pipeline(
    createReadStream('orders.csv'),
    createGzip(),
    createWriteStream('orders.csv.gz'),
    { signal: AbortSignal.timeout(500) }, // Abbruch nach 500 ms
  );
} catch (err) {
  console.error(err.name); // AbortError bei Timeout
}

AbortError

Wenn es abgebrochen wird, werden alle Streams geschlossen. Die Datei orders.csv.gz ist dann nur teilweise geschrieben. In echtem Code solltest du diese Datei im catch-Block löschen oder in eine temporäre Datei schreiben und sie nach Abschluss umbenennen.

7. Transform testen

Die fehleranfälligste Logik ist das Zusammensetzen von Zeilen, die zwischen Chunks aufgeteilt sind. Lass uns den integrierten Test-Runner node:test nutzen. Die Mock-Datenquelle ist Readable.from mit Chunks, die absichtlich Zeilen und UTF-8-Zeichen durchschneiden:

// split-lines.test.mjs
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { Readable } from 'node:stream';
import { splitLines } from './split-lines.mjs';

async function collect(stream) {
  const result = [];
  for await (const item of stream) result.push(item);
  return result;
}

test('Zeilen korrekt zusammensetzen, die zwischen Chunks aufgeteilt sind', async () => {
  const source = Readable.from([
    Buffer.from('1,an,10,PAID\n2,bi'),
    Buffer.from('nh,20,FAILED\n3,chi,5,PAID'), // letzte Zeile hat kein \n
  ]);

  const lines = await collect(source.pipe(splitLines()));

  assert.deepEqual(lines, ['1,an,10,PAID', '2,binh,20,FAILED', '3,chi,5,PAID']);
});

test('Mehrbyte UTF-8-Zeichen nicht beschädigen', async () => {
  const bytes = Buffer.from('1,Nguyễn Văn Hiệp,10,PAID\n');
  const source = Readable.from([bytes.subarray(0, 9), bytes.subarray(9)]); // schneidet das Zeichen "ễ" durch

  const lines = await collect(source.pipe(splitLines()));

  assert.deepEqual(lines, ['1,Nguyễn Văn Hiệp,10,PAID']);
});

Test ausführen:

node --test

✔ Zeilen korrekt zusammensetzen, die zwischen Chunks aufgeteilt sind (3.1ms)
✔ Mehrbyte UTF-8-Zeichen nicht beschädigen (0.6ms)
ℹ tests 2
ℹ pass 2
ℹ fail 0

In diesem Test ist .pipe() in Ordnung, weil Readable.from im Speicher läuft und keine I/O-Ressourcen aufgeräumt werden müssen. Wenn du StringDecoder weglässt und chunk.toString() direkt aufrufst, schlägt der zweite Test fehl mit dem fehlerhaften Zeichen �.

8. Häufige Fehler und wann man Streams nicht nutzen sollte

on('data') mit async-Handler nutzen. Zum Beispiel stream.on('data', async chunk => await db.insert(chunk)). Der Event Emitter wartet nicht auf das Promise, daher schiebt der Stream weiterhin Daten mit maximaler Geschwindigkeit, und tausende insert-Befehle laufen parallel. Nutze stattdessen for await oder einen Async-Generator-Schritt in pipeline.

Fehlerbehandlung vergessen. Ein Stream, der 'error' emittiert, ohne einen Listener zu haben, wird den ganzen Prozess zum Absturz bringen. Mit dem Promise-basierten pipeline brauchst du nur eine try/catch.

Annahme, dass ein Chunk eine Zeile oder einen Datensatz darstellt. Chunks werden nach Bytes aufgeteilt. Trenne Zeilen immer mit readline oder einem Transform mit einem tail-Buffer wie in Abschnitt 6.2.

highWaterMark zu groß setzen. Wenn du 100 MB setzt, um “schneller zu lesen”, hilft das kaum, sondern zerstört nur die Speichereffizienz. Werte zwischen 64 KiB und 1 MiB sind normalerweise ausreichend.

Ergebnisse in einem großen Array sammeln. Code wie for await (...) rows.push(row) und dann verarbeiten führt in das gleiche Problem wie readFile. Verarbeite oder schreibe die Daten sofort nach dem Empfang.

Hinweis: Streams sind nicht immer die beste Wahl. Bei kleinen Konfigurationsdateien oder wenn du alle Daten gleichzeitig brauchst (z. B. zum Sortieren oder Parsen einer JSON-Datei), ist readFile einfacher und schneller. Nutze Streams nur bei großen Datenmengen, unbekannter Größe oder Daten aus dem Netzwerk.

9. Fazit

Du hast gesehen, dass readFile viel mehr RAM verbraucht als die Dateigröße. Streams halten den RAM-Verbrauch stabil, weil sie Daten Chunk für Chunk verarbeiten. Wichtige Punkte:

• Nutze createReadStream mit for await, um Chunks zu lesen, und readline für zeilenweises Lesen.
• Beachte Backpressure: Prüfe den Rückgabewert von write(), oder lass pipeline das übernehmen.
• Nutze pipeline aus node:stream/promises statt .pipe(), damit Fehler durch alle Streams weitergeleitet werden und Ressourcen korrekt aufgeräumt werden.
• Schreibe Transform-Klassen mit Behandlung für unvollständige Zeilen und Bytes, und teste sie mit absichtlich durchgeschnittenen Chunks.

Mit diesen integrierten Tools kannst du mehrere Gigabyte große Dateien filtern, umwandeln und komprimieren, während der Prozess nur wenige Dutzend Megabyte RAM verbraucht.