Stream dans Node.js : traiter les gros fichiers sans épuiser la RAM

1. Aperçu

Lire un fichier entier en mémoire avec fs.readFile est pratique pour les petits fichiers. Pour des fichiers log de plusieurs GB ou des fichiers CSV avec des millions de lignes, cette approche fait gonfler la RAM du process ou le fait crash. Cet article vous guide pour traiter les gros fichiers avec les streams dans Node.js : lire par chunk, séparer les lignes, filtrer les données, compresser et écrire dans un nouveau fichier. Vous comprendrez aussi le backpressure et comment gérer les erreurs correctement avec pipeline.

Tous les exemples utilisent Node.js 20+ et les modules ES. Le fichier package.json a besoin d’avoir "type": "module". L’article ne nécessite aucune bibliothèque externe car fs, stream, zlib et readline sont intégrés.

2. Le problème avec readFile

D’abord, créons un fichier orders.csv d’environ 5 millions de lignes pour expérimenter :

// 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'); // attendre l'événement 'drain' du buffer d'écriture
}
out.end();

Ce code utilise déjà le backpressure. La section 5 expliquera ce mécanisme en détail. Le fichier créé pèse environ 180 MB.

Maintenant, comptons le nombre de commandes FAILED de la manière « naïve » :

// 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

Un fichier de 180 MB consomme environ 1 GB de RAM. Le process doit conserver simultanément trois choses : le buffer brut, la chaîne UTF-8 et un tableau de 5 millions de sous-chaînes. Quand le fichier dépasse environ 2 GB, readFile lève ERR_FS_FILE_TOO_LARGE et ne peut plus le lire.

3. Les quatre types de stream de base

Un stream est des données qui passent par morceaux (chunks). Node.js a quatre types de stream :

Type	Rôle	Exemple
Readable	Source qui émet des données	fs.createReadStream, process.stdin
Writable	Destination qui reçoit les données	fs.createWriteStream, process.stdout
Duplex	À la fois lecture et écriture, deux sens indépendants	Socket TCP
Transform	Duplex dont la sortie dépend de l’entrée	zlib.createGzip()

Chaque stream a un buffer interne. La limite de ce buffer est déterminée par highWaterMark. Par défaut, fs.createReadStream utilise 64 KiB. Les streams en mode objet utilisent 16 objets.

4. Lire un gros fichier par chunk

4.1. Utiliser for await sur un Readable

La manière la plus concise pour consommer un Readable est l’itération asynchrone :

// 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

Chaque chunk est un Buffer d’au maximum 64 KiB. La boucle demande le chunk suivant seulement après avoir traité le chunk actuel. Donc la RAM ne dépend pratiquement pas de la taille du fichier.

4.2. Lire ligne par ligne avec readline

Un chunk coupe par octet, pas par ligne. Pour traiter chaque ligne, utilisez readline :

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

const rl = readline.createInterface({
  input: createReadStream('orders.csv'),
  crlfDelay: Infinity, // traiter \r\n comme une seule rupture de ligne
});

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

FAILED: 500000

Le résultat est identique à la section 2 mais la RAM n’est que 50–60 MB. readline assemble automatiquement les parties de ligne coupées entre deux chunks, donc vous n’avez pas à vous en préoccuper.

5. Écrire un fichier et backpressure

Le backpressure se produit quand la source émet des données plus vite que la destination ne peut les recevoir. Un disque, un réseau ou gzip peuvent tous être plus lents que la vitesse de lecture. La fonction writable.write() retourne false quand le buffer interne dépasse highWaterMark. C’est un signal : « arrête-toi, attends l’événement drain ».

L’exemple suivant ignore ce signal :

// bad-write.mjs — NE PAS faire cela
import { createWriteStream } from 'node:fs';

const out = createWriteStream('big.txt');
for (let i = 0; i < 10_000_000; i++) {
  out.write(`line ${i}\n`); // ignorer la valeur de retour
}
out.end();

La boucle s’exécute de manière synchrone, donc les 10 millions de chaînes s’accumulent dans le buffer de out avant que le disque ne puisse les écrire. La RAM bondit à plusieurs GB. La solution est vérifier la valeur de retour puis await once(out, 'drain'), comme dans le fichier generate-orders.mjs de la section 2.

Si vous enchaînez les streams avec pipeline, le backpressure est géré automatiquement. C’est la façon à utiliser dans la plupart des cas.

6. Enchaîner les streams avec pipeline

6.1. Copier et compresser un fichier

// 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 provenant de node:stream/promises retourne une Promise. Cette Promise se résout quand le dernier stream a fini d’écrire. Si un stream quelconque échoue, par exemple le fichier n’existe pas ou le disque est plein, la Promise est rejetée. À ce moment, tous les streams sont détruits, donc les file descriptors ne fuient pas.

Note : Évitez a.pipe(b).pipe(c). pipe ne propage pas les erreurs à travers les streams. Si a échoue, c reste ouvert indéfiniment. Utilisez pipeline à la place.

6.2. Écrire un Transform qui sépare les lignes

Supposons que vous vouliez filtrer les commandes FAILED puis les compresser. Vous avez besoin d’une étape qui sépare les lignes et qui peut être mise dans pipeline. Écrivons notre propre 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, // la sortie produit des lignes individuelles (string), pas des Buffer

    transform(chunk, _encoding, callback) {
      const lines = (tail + decoder.write(chunk)).split('\n');
      tail = lines.pop(); // la fin pourrait être une ligne incomplète
      for (const line of lines) this.push(line);
      callback();
    },

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

Explication de chaque partie :

• readableObjectMode: true : le côté lecture émet chaque ligne comme un objet distinct. Le côté écriture reçoit toujours des Buffer.
• StringDecoder : un chunk peut couper au milieu d’un caractère UTF-8 multi-octet, par exemple la lettre « ệ » dans « Nguyễn Văn Hiệp ». Appeler chunk.toString() directement créerait un caractère corrompu �. decoder.write() conserve les octets incomplets et les assemble au prochain appel.
• tail : stocke la partie de ligne incomplète pour la concaténer avec le chunk suivant.
• flush : s’exécute quand l’entrée se termine. Cette étape pousse la dernière ligne si le fichier n’a pas de \n à la fin.

6.3. Filtrer avec un générateur asynchrone

pipeline accepte un générateur asynchrone comme une étape. Grâce à cela, vous pouvez filtrer sans avoir à écrire une classe Transform supplémentaire :

// 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(`Exported ${count} failed orders`);

Exported 500000 failed orders

Les données s’écoulent à travers cinq étapes. Chaque étape conserve seulement quelques chunks en mémoire. Quand gzip est lent, le générateur se suspend au yield et splitLines arrête de recevoir des chunks. En résultat, createReadStream met en pause la lecture du disque. C’est exactement ce qu’est le backpressure fonctionnant automatiquement.

6.4. Annuler à mi-parcours avec AbortSignal

Avec des tâches longues, vous devriez permettre l’annulation ou définir un timeout :

// 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) }, // annuler après 500 ms
  );
} catch (err) {
  console.error(err.name); // AbortError si dépassement du délai
}

AbortError

Quand annulé, tous les streams sont fermés. Le fichier orders.csv.gz à ce moment n’est écrit qu’en partie. Dans le code réel, supprimez ce fichier dans le bloc catch, ou écrivez dans un fichier temporaire puis renommez-le une fois terminé.

7. Tester un Transform

La logique la plus facile à mal comprendre est l’assemblage des lignes coupées entre deux chunks. Testons avec le test runner intégré node:test. La source de données factice est Readable.from avec des chunks volontairement coupés au milieu d’une ligne et au milieu d’un caractère UTF-8 :

// 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('concaténation des lignes coupées entre les chunks', async () => {
  const source = Readable.from([
    Buffer.from('1,an,10,PAID\n2,bi'),
    Buffer.from('nh,20,FAILED\n3,chi,5,PAID'), // dernière ligne sans \n
  ]);

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

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

test('ne pas corrompre les caractères UTF-8 multi-octets', 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)]); // couper au milieu de la lettre « ệ »

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

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

Lancer les tests :

node --test

✔ concaténation des lignes coupées entre les chunks (3.1ms)
✔ ne pas corrompre les caractères UTF-8 multi-octets (0.6ms)
ℹ tests 2
ℹ pass 2
ℹ fail 0

Dans ce test, utiliser .pipe() c’est ok car Readable.from s’exécute en mémoire et n’a pas d’I/O à nettoyer. Si vous supprimez StringDecoder et utilisez chunk.toString(), le deuxième test échouera avec une chaîne contenant �.

8. Erreurs courantes et quand ne pas utiliser les streams

Utiliser on('data') avec un gestionnaire async. Par exemple stream.on('data', async chunk => await db.insert(chunk)). L’émetteur d’événement n’attend pas les Promises, donc le stream continue à pousser les données à pleine vitesse et des milliers d’insert s’exécutent en parallèle. Utilisez for await ou une étape de générateur asynchrone dans pipeline.

Oublier de gérer les erreurs. Un stream qui émet 'error' sans listener va crash l’ensemble du process. Avec pipeline sous forme de Promise, vous avez juste besoin de try/catch à un seul endroit.

Supposer qu’un chunk est une ligne ou un enregistrement. Les chunks sont coupés par octet. Séparez toujours les lignes avec readline ou un Transform avec un buffer tail comme dans la section 6.2.

Définir highWaterMark trop grand. Mettre 100 MB pour « lire plus vite » n’aide presque pas, et perd les bénéfices de mémoire. Une valeur de 64 KiB à 1 MiB est généralement suffisante.

Accumuler le résultat dans un grand tableau. Le code du type for await (...) rows.push(row) puis traiter après revient exactement au problème de readFile. Traitez ou écrivez les données immédiatement une fois reçues.

Note : Les streams ne sont pas toujours le meilleur choix. Pour un fichier de configuration de quelques KB, ou quand vous avez besoin de toutes les données à la fois (trier tout, parser un seul fichier JSON), readFile est plus simple et plus rapide. Utilisez les streams seulement quand la taille des données est grande, inconnue d’avance, ou provient du réseau.

9. Conclusion

Vous avez vu que readFile consomme plusieurs fois plus de RAM que la taille du fichier. Les streams gardent la mémoire stable en traitant chaque chunk. Les points clés :

• Utilisez createReadStream avec for await pour lire par chunk, et readline pour lire par ligne.
• Respectez le backpressure : vérifiez la valeur de retour de write(), ou laissez pipeline s’en charger.
• Utilisez pipeline depuis node:stream/promises à la place de .pipe() pour que les erreurs se propagent à travers les streams et les ressources soient nettoyées correctement.
• Écrivez des Transform qui gèrent les lignes incomplètes et les octets UTF-8 incomplets, puis testez avec des chunks volontairement coupés.

Avec ces outils intégrés, vous pouvez filtrer, transformer et compresser des fichiers de plusieurs GB tout en ne consommant que quelques dizaines de MB de RAM.