Stream en Node.js: procesar archivos grandes sin agotar la memoria

1. Descripción general

Leer todo un archivo en memoria usando fs.readFile es conveniente cuando el archivo es pequeño. Con logs de algunos GB o archivos CSV de millones de líneas, este enfoque hace que el proceso consuma más RAM o se bloquee. Este artículo te guía en el procesamiento de archivos grandes usando streams en Node.js: leer cada chunk, dividir líneas, filtrar datos, comprimir y escribir en un archivo nuevo. También entenderás qué es backpressure y cómo manejar errores correctamente con pipeline.

Todos los ejemplos usan Node.js 20+ y ES module. El archivo package.json necesita "type": "module". Este artículo no requiere librerías externas ya que fs, stream, zlib y readline vienen incluidas.

2. El problema con readFile

Primero, creamos un archivo orders.csv de alrededor de 5 millones de líneas para hacer pruebas:

// 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'); // esperar a que el buffer de escritura se vacíe antes de escribir más
}
out.end();

Este código ya usa backpressure. La sección 5 explicará el mecanismo en detalle. El archivo generado pesa alrededor de 180 MB.

Ahora contemos el número de pedidos FAILED de forma “ingenua”:

// 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 archivo de 180 MB consumió alrededor de 1 GB de RAM. El proceso debe mantener simultáneamente tres cosas: el buffer original, la cadena UTF-8 y un array de 5 millones de subcadenas. Cuando el archivo supera aproximadamente 2 GB, readFile lanzará ERR_FS_FILE_TOO_LARGE y no podrá leer más.

3. Cuatro tipos de stream básicos

Un stream es datos que fluyen en trozos (chunks). Node.js tiene cuatro tipos de stream:

Tipo	Propósito	Ejemplo
Readable	Fuente que emite datos	fs.createReadStream, process.stdin
Writable	Destino que recibe datos	fs.createWriteStream, process.stdout
Duplex	Lee y escribe, dos direcciones independientes	TCP socket
Transform	Duplex cuya salida depende de la entrada	zlib.createGzip()

Cada stream tiene un buffer interno. El límite de este buffer se determina por highWaterMark. Por defecto, fs.createReadStream usa 64 KiB. Los streams en object mode usan 16 objetos.

4. Leer archivo grande por chunks

4.1. Usar for await en Readable

La forma más concisa de consumir un Readable es la iteración asincrónica:

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

Cada chunk es un Buffer de máximo 64 KiB. El bucle solo solicita el siguiente chunk después de procesar el actual. Por lo tanto, la RAM es prácticamente independiente del tamaño del archivo.

4.2. Leer línea por línea con readline

Los chunks se dividen por bytes, no por líneas. Si necesitas procesar línea por línea, usa readline:

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

const rl = readline.createInterface({
  input: createReadStream('orders.csv'),
  crlfDelay: Infinity, // tratar \r\n como un salto de línea
});

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

FAILED: 500000

El resultado es el mismo que en la sección 2, pero la RAM utilizada es solo de 50–60 MB. readline automáticamente une las líneas que se cortaron entre dos chunks, así que no tienes que preocuparte por eso.

5. Escribir archivo y backpressure

El backpressure ocurre cuando la fuente emite datos más rápido de lo que el destino puede escribir. El disco, la red o gzip pueden ser más lentos que la velocidad de lectura. La función writable.write() devuelve false cuando el buffer interno supera highWaterMark. Esa es la señal de “detente, espera el evento drain”.

El siguiente ejemplo ignora esta señal:

// bad-write.mjs — NO hagas esto
import { createWriteStream } from 'node:fs';

const out = createWriteStream('big.txt');
for (let i = 0; i < 10_000_000; i++) {
  out.write(`line ${i}\n`); // ignora el valor de retorno
}
out.end();

El bucle es síncrono, por lo que los 10 millones de cadenas se acumulan en el buffer de out antes de que el disco pueda escribirlas. La RAM se dispara a varios GB. La forma de arreglarlo es verificar el valor de retorno y luego await once(out, 'drain'), como en el archivo generate-orders.mjs de la sección 2.

Si encadenas streams usando pipeline, el backpressure se maneja automáticamente. Esta es la forma que debes usar en la mayoría de casos.

6. Encadenar streams con pipeline

6.1. Copiar y comprimir archivo

// 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 de node:stream/promises devuelve una Promise. Esta Promise se resuelve cuando el último stream termina de escribir. Si algún stream falla, por ejemplo si el archivo no existe o el disco está lleno, la Promise se rechaza. En ese caso, todos los streams se destruyen, por lo que los descriptores de archivo no tienen fugas.

Nota: Evita usar a.pipe(b).pipe(c). pipe no propaga errores a través de los streams. Si a falla, c seguirá abierto indefinidamente. Usa pipeline en su lugar.

6.2. Escribir un Transform para dividir líneas

Supongamos que quieres filtrar pedidos FAILED y luego comprimirlos. Necesitas un paso que divida líneas que puedas poner en pipeline. Escribamos nuestro propio 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 salida es línea por línea (string), no Buffer

    transform(chunk, _encoding, callback) {
      const lines = (tail + decoder.write(chunk)).split('\n');
      tail = lines.pop(); // la última parte podría ser una línea incompleta
      for (const line of lines) this.push(line);
      callback();
    },

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

Explicación de cada parte:

• readableObjectMode: true: el lado de lectura emite cada línea como un objeto separado. El lado de escritura aún recibe Buffer.
• StringDecoder: un chunk puede cortarse en medio de un carácter UTF-8 de varios bytes, como la letra “ễ” en “Nguyễn Văn Hiệp”. Llamar a chunk.toString() directamente producirá un carácter corrupto �. decoder.write() conserva los bytes incompletos y los une en la siguiente llamada.
• tail: almacena la parte de la línea sin terminar para unirla con el siguiente chunk.
• flush: se ejecuta cuando la entrada termina. Este paso envía la última línea si el archivo no tiene \n al final.

6.3. Filtrar usando async generator

pipeline acepta un async generator como un paso. Gracias a eso, puedes filtrar sin necesidad de escribir una clase Transform adicional:

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

Los datos fluyen a través de cinco pasos. Cada paso mantiene solo unos pocos chunks en memoria. Cuando gzip es lento, el generator se detiene en yield y splitLines deja de recibir chunks. Como resultado, createReadStream pausa temporalmente la lectura del disco. Ese es el backpressure ejecutándose automáticamente.

6.4. Cancelar a mitad de camino con AbortSignal

Para tareas largas, debes permitir la cancelación o establecer 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) }, // cancelar después de 500 ms
  );
} catch (err) {
  console.error(err.name); // AbortError si se excede el tiempo
}

AbortError

Cuando se cancela, todos los streams se cierran. El archivo orders.csv.gz en este punto solo está escrito parcialmente. En código real, debes eliminar ese archivo en el bloque catch, o escribir a un archivo temporal y hacer rename cuando termines.

7. Prueba el Transform

La lógica más propensa a errores es el manejo de líneas cortadas entre chunks. Probamos usando el test runner integrado node:test. La fuente de datos falsa es Readable.from con chunks que deliberadamente cortan líneas y caracteres 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('une líneas cortadas entre chunks', async () => {
  const source = Readable.from([
    Buffer.from('1,an,10,PAID\n2,bi'),
    Buffer.from('nh,20,FAILED\n3,chi,5,PAID'), // última línea sin \n
  ]);

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

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

test('no corrompe caracteres UTF-8 de varios bytes', 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)]); // corta en medio de "ễ"

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

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

Ejecuta las pruebas:

node --test

✔ une líneas cortadas entre chunks (3.1ms)
✔ no corrompe caracteres UTF-8 de varios bytes (0.6ms)
ℹ tests 2
ℹ pass 2
ℹ fail 0

En este test, usar .pipe() está bien porque Readable.from se ejecuta en memoria y no hay I/O que limpiar. Si eliminas StringDecoder y usas chunk.toString() directamente, la segunda prueba fallará con una cadena que contiene �.

8. Errores comunes y cuándo no usar streams

Usar on('data') con un handler async. Por ejemplo stream.on('data', async chunk => await db.insert(chunk)). El emisor de eventos no espera la Promise, así que el stream sigue emitiendo datos a velocidad máxima y miles de insert se ejecutan en paralelo. Usa for await o un async generator dentro de pipeline.

Olvidar manejar errores. Un stream que emite 'error' sin un listener causará que todo el proceso falle. Con pipeline en forma de Promise, solo necesitas un try/catch en un lugar.

Asumir que un chunk es una línea o un registro. Los chunks se cortan por bytes. Siempre divide las líneas usando readline o un Transform con buffer tail como en la sección 6.2.

Establecer highWaterMark muy grande. Establecerlo a 100 MB para “leer más rápido” casi no ayuda, y pierdes los beneficios de memoria. Valores entre 64 KiB y 1 MiB suelen ser suficientes.

Acumular resultados en un array grande. Código como for await (...) rows.push(row) y luego procesarlos te devuelve exactamente al problema de readFile. Procesa o escribe los datos inmediatamente cuando los recibas.

Nota: Los streams no siempre son la mejor opción. Para archivos de configuración de unos pocos KB, o cuando necesitas todos los datos a la vez (ordenar todo, parsear un JSON simple), readFile es más simple y rápido. Usa streams solo cuando el volumen de datos es grande, no se conoce de antemano, o proviene de la red.

9. Conclusión

Has visto cómo readFile consume RAM varias veces el tamaño del archivo. Los streams mantienen la memoria estable procesando cada chunk. Los puntos principales:

• Usa createReadStream con for await para leer por chunks, y readline para leer línea por línea.
• Respeta el backpressure: verifica el valor de retorno de write(), o deja que pipeline se encargue.
• Usa pipeline de node:stream/promises en lugar de .pipe() para que los errores se propaguen a través de los streams y los recursos se limpien correctamente.
• Escribe Transform que maneje líneas incompletas y bytes UTF-8 incompletos, luego prueba con chunks que deliberadamente corten por la mitad.

Con estas herramientas integradas, puedes filtrar, transformar y comprimir archivos de muchos GB mientras el proceso solo usa decenas de MB de RAM.