backend / nodejs

Fichiers et streams

Leçon 31 exercice

Explication

Ce que vous allez apprendre

Comprendre pourquoi readFile charge tout un fichier en mémoire d'un coup
Lire et écrire un fichier volumineux chunk par chunk avec des streams
Enchaîner plusieurs streams proprement avec pipeline() plutôt que .pipe()
Créer un Transform stream personnalisé pour traiter un flux de données
Éviter le piège classique du callback() oublié dans _transform()

Dans quel contexte ?

Un service doit exporter un fichier CSV de plusieurs millions de lignes vers un client, ou compresser un fichier de logs de 2 Go avant de l'archiver. Charger ce fichier entièrement avec readFile avant de le traiter ferait exploser la mémoire du processus Node, voire planter le serveur entier. Cette leçon montre comment traiter ce genre de volume par petits morceaux, sans jamais garder tout le fichier en mémoire en même temps.

Le problème, d'abord

Imagine devoir déplacer une piscine d'eau avec un seul grand seau. Tu dois attendre de le remplir ENTIÈREMENT avant de pouvoir le vider ailleurs — c'est exactement ce que fait readFile avec un fichier.

Pour un petit fichier JSON de config, aucun souci. Mais pour un export CSV de plusieurs gigaoctets, charger tout le fichier en mémoire d'un coup peut faire planter le processus ou simplement le ralentir énormément.

Les streams résolvent ça avec une idée simple : traiter par petits morceaux. Un stream, c'est comme un tuyau : les données y circulent par petits "chunks", traités au fur et à mesure de leur arrivée, sans jamais avoir besoin de tout garder en mémoire en même temps.

Une image aide à retenir le principe. C'est le même principe que regarder une vidéo en streaming plutôt que de télécharger tout le film avant de pouvoir appuyer sur "lecture".

Une fois qu'on sait lire par morceaux, une question se pose : comment enchaîner plusieurs étapes ? Par exemple lire un fichier, le compresser, puis l'écrire ailleurs. La méthode .pipe() permet ça, mais gère mal les erreurs.

Voici pourquoi : si une erreur survient au milieu de la chaîne avec .pipe(), les streams précédents peuvent rester ouverts. C'est une fuite de ressources difficile à repérer.

Méthode	Gestion des erreurs	Fermeture des streams
`.pipe()`	Manuelle, facile à oublier	Pas automatique en cas d'erreur
`pipeline()`	Propagée automatiquement	Tous les streams sont fermés proprement

Prérequis

Cette leçon suppose que tu es à l'aise avec les modules CommonJS/ESM vus précédemment : les exemples utilisent import pour charger les modules natifs node:fs et node:stream.

pipeline() règle ce problème précis. Il propage automatiquement les erreurs ET ferme proprement chaque stream impliqué — c'est la méthode recommandée en pratique, pas juste un détail de style.

Il reste un dernier outil à connaître : les Transform streams. Ils permettent d'insérer une étape de traitement entre la lecture et l'écriture — filtrer des lignes, convertir un format, chiffrer des données — sans jamais matérialiser le résultat intermédiaire complet en mémoire.

Le piège fréquent à connaître avant de pratiquer : oublier d'appeler callback() à la fin de _transform() bloque tout le flux, car le stream attend indéfiniment ce signal pour savoir qu'il peut traiter le chunk suivant.

Piège fréquent

Dans un Transform custom, oublier d'appeler callback() (ou callback(err) en cas d'erreur) à la fin de _transform() fait que le stream reste bloqué indéfiniment, sans message d'erreur explicite. C'est l'un des bugs les plus difficiles à diagnostiquer sur des streams personnalisés.

Et la suite ? Cette leçon prolonge la précédente sur les modules : comprendre comment organiser le code va maintenant de pair avec comprendre comment le faire passer à l'échelle sur de gros volumes de données.

Commandes & code

Fichiers et streams

// API Promise — lecture/écriture simple, tout en mémoire
import { readFile, writeFile, appendFile, mkdir } from "node:fs/promises";

async function main() {
  await mkdir("./logs", { recursive: true });

  const content = await readFile("./data.json", "utf-8");
  const data = JSON.parse(content);

  data.updatedAt = new Date().toISOString();
  await writeFile("./data.json", JSON.stringify(data, null, 2));

  await appendFile("./logs/app.log", `[${new Date().toISOString()}] Fichier mis à jour\n`);
}

main().catch(console.error);

// Streams — traiter un fichier volumineux sans le charger entièrement en RAM
import { createReadStream, createWriteStream } from "node:fs";

const readStream = createReadStream("./huge-export.csv", { encoding: "utf-8" });
const writeStream = createWriteStream("./huge-export-filtered.csv");

let buffer = "";
let isFirstLine = true;

readStream.on("data", (chunk) => {
  buffer += chunk;
  const lines = buffer.split("\n");
  buffer = lines.pop() ?? ""; // garde la ligne incomplète pour le prochain chunk

  for (const line of lines) {
    if (isFirstLine) {
      writeStream.write(line + "\n");
      isFirstLine = false;
      continue;
    }
    if (line.includes(",active,")) {
      writeStream.write(line + "\n");
    }
  }
});

readStream.on("end", () => {
  writeStream.end();
  console.log("Traitement terminé");
});

readStream.on("error", (err) => console.error("Erreur de lecture:", err));

// pipeline() — enchaîne des streams avec gestion propre des erreurs et du cleanup
import { pipeline } from "node:stream/promises";
import { createReadStream, createWriteStream } from "node:fs";
import { createGzip } from "node:zlib";

async function compressFile(inputPath, outputPath) {
  await pipeline(
    createReadStream(inputPath),
    createGzip(),
    createWriteStream(outputPath)
  );
  console.log(`${inputPath} compressé vers ${outputPath}`);
}

await compressFile("./access.log", "./access.log.gz");

// Transform stream custom — traiter un flux de données ligne par ligne
import { Transform } from "node:stream";

class UpperCaseTransform extends Transform {
  _transform(chunk, encoding, callback) {
    this.push(chunk.toString().toUpperCase());
    callback(); // signale que ce chunk est traité, permet au flux de continuer
  }
}

process.stdin.pipe(new UpperCaseTransform()).pipe(process.stdout);

// Watcher de fichiers — utile pour le hot-reload ou la synchronisation
import { watch } from "node:fs";

const watcher = watch("./config", { recursive: true }, (eventType, filename) => {
  console.log(`Événement "${eventType}" sur ${filename}`);
  // recharger la config, invalider un cache, etc.
});

process.on("SIGINT", () => {
  watcher.close();
  process.exit(0);
});

Résumé

fs/promises convient pour des fichiers de taille raisonnable, chargés entièrement en mémoire.
Les streams traitent des fichiers volumineux chunk par chunk, sans exploser la mémoire.
pipeline() gère automatiquement la propagation d'erreurs et le cleanup (contrairement à .pipe() seul).
Un Transform stream custom permet d'insérer une étape de traitement dans une chaîne de streams.

Exercices pratiques

1 disponible

Mission : exporter un CSV de 3 millions de lignes sans faire planter le serveur

Objectif : Remplacer un traitement readFile qui sature la mémoire par une chaîne de streams robuste avec pipeline(), et corriger un Transform custom qui bloque le flux.

Contexte

Un endpoint d'export utilise actuellement const content = await readFile('./export.csv', 'utf-8') avant de filtrer les lignes actives et de compresser le résultat. Sur un fichier de test de 3 millions de lignes (environ 1,5 Go), le processus Node consomme plus de 4 Go de RAM et finit par crasher avec JavaScript heap out of memory. Un collègue a commencé à réécrire un Transform custom pour filtrer les lignes, mais le flux reste bloqué indéfiniment après le premier chunk.

Résoudre l’exercice →