Stream no Node.js: processando arquivos grandes sem esgotar memória

1. Visão geral

Ler todo um arquivo na memória com fs.readFile é conveniente quando o arquivo é pequeno. Com arquivos de log de alguns GB ou arquivos CSV com milhões de linhas, essa abordagem faz o processo inchar a RAM ou travar. Este artigo vai guiá-lo no processamento de arquivos grandes usando streams no Node.js: ler por chunk, separar linhas, filtrar dados, comprimir e gravar em um novo arquivo. Você também aprenderá sobre backpressure e como lidar com erros corretamente com pipeline.

Todos os exemplos usam Node.js 20+ e módulos ES. O arquivo package.json precisa ter "type": "module". Este artigo não requer bibliotecas externas, pois fs, stream, zlib e readline vêm incluídos.

2. O problema com readFile

Primeiro, vamos criar um arquivo orders.csv com cerca de 5 milhões de linhas para teste:

// 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'); // aguarde o buffer gravar antes de continuar
}
out.end();

Este trecho já usa backpressure. A seção 5 explicará esse mecanismo em detalhes. O arquivo gerado pesa cerca de 180 MB.

Agora vamos contar o número de pedidos FAILED da forma “ingênua”:

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

Um arquivo de 180 MB consumiu cerca de 1 GB de RAM. O processo precisa manter simultaneamente três coisas: o buffer original, a string UTF-8 e um array de 5 milhões de substrings. Quando o arquivo é maior que aproximadamente 2 GB, readFile lançará ERR_FS_FILE_TOO_LARGE e não conseguirá ler.

3. Quatro tipos de stream básicos

Um stream é dados que fluem em pedaços (chunks). Node.js tem quatro tipos de stream:

Tipo	Função	Exemplo
Readable	Fonte de dados	fs.createReadStream, process.stdin
Writable	Destino de dados	fs.createWriteStream, process.stdout
Duplex	Lê e escreve, dois sentidos independentes	TCP socket
Transform	Duplex com saída dependente da entrada	zlib.createGzip()

Cada stream tem um buffer interno. O limite desse buffer é determinado por highWaterMark. Por padrão, fs.createReadStream usa 64 KiB. Streams em object mode usam 16 objetos.

4. Lendo arquivos grandes por chunk

4.1. Usando for await em Readable

A forma mais concisa de consumir um Readable é através de iteração assíncrona:

// 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 é um Buffer de no máximo 64 KiB. O loop apenas solicita o próximo chunk após processar o atual. Portanto, a RAM é quase independente do tamanho do arquivo.

4.2. Lendo linha por linha com readline

Chunks são cortados por byte, não por linha. Se você precisar processar linha por linha, use readline:

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

const rl = readline.createInterface({
  input: createReadStream('orders.csv'),
  crlfDelay: Infinity, // considere \r\n como uma única quebra de linha
});

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

FAILED: 500000

O resultado é o mesmo que na seção 2, mas a RSS fica em torno de 50–60 MB. readline automaticamente emenda as linhas cortadas entre dois chunks, então você não precisa se preocupar com isso.

5. Escrevendo arquivos e backpressure

Backpressure ocorre quando a fonte de dados é mais rápida que o destino de escrita. Disco rígido, rede ou gzip podem ser mais lentos que a velocidade de leitura. A função writable.write() retorna false quando o buffer interno ultrapassa highWaterMark. É um sinal para “parar, aguarde o evento drain”.

O exemplo a seguir ignora esse sinal:

// bad-write.mjs — NÃO faça assim
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 o valor retornado
}
out.end();

O loop é executado de forma síncrona, então todas as 10 milhões de strings são enfileiradas no buffer de out antes que o disco consiga escrever. A RAM sobe para vários GB. A correção é verificar o valor retornado e então await once(out, 'drain'), como no arquivo generate-orders.mjs da seção 2.

Se você conectar streams usando pipeline, o backpressure é tratado automaticamente. Esta é a forma como deve ser feito na maioria dos casos.

6. Conectando streams com pipeline

6.1. Copiando e comprimindo arquivo

// 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 do node:stream/promises retorna uma Promise. Esta Promise resolve quando o último stream termina de escrever. Se qualquer stream der erro, por exemplo, se o arquivo não existir ou o disco estiver cheio, a Promise será rejeitada. Quando isso acontece, todos os streams são destroyidos, então os file descriptors não vazam.

Nota: Evite usar a.pipe(b).pipe(c). pipe não propaga erros entre streams. Se a falhar, c permanece aberto indefinidamente. Use pipeline em seu lugar.

6.2. Escrevendo Transform para separar linhas

Suponha que você quer filtrar pedidos FAILED e depois compactá-los. Você precisa de uma etapa para separar linhas que possa ser inserida em pipeline. Vamos escrever um 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, // a saída são linhas individuais (string), não Buffer

    transform(chunk, _encoding, callback) {
      const lines = (tail + decoder.write(chunk)).split('\n');
      tail = lines.pop(); // a última parte pode ser uma linha incompleta
      for (const line of lines) this.push(line);
      callback();
    },

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

Explicação de cada parte:

• readableObjectMode: true: o lado legível emite cada linha como um objeto separado. O lado gravável continua recebendo Buffer.
• StringDecoder: um chunk pode ser cortado no meio de um caractere UTF-8 multibyte, como a letra “ễ” em “Nguyễn Văn Hiệp”. Chamar chunk.toString() diretamente produzirá um caractere de erro �. decoder.write() retém bytes incompletos e os cola na próxima chamada.
• tail: armazena a parte da linha incompleta para emendá-la com o próximo chunk.
• flush: executado quando a entrada termina. Esta etapa emite a linha final se o arquivo não terminar com \n.

6.3. Filtrando com async generator

pipeline aceita um async generator como uma etapa. Assim você pode filtrar sem escrever uma classe 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

Os dados fluem por cinco etapas. Cada etapa mantém apenas alguns chunks na memória. Quando gzip é lento, o generator pausa no yield e splitLines para de receber chunks. Como resultado, createReadStream pausa a leitura do disco. Isso é backpressure funcionando automaticamente.

6.4. Cancelando no meio com AbortSignal

Para tarefas longas, você deve permitir cancelamento ou definir um 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) }, // cancele após 500 ms
  );
} catch (err) {
  console.error(err.name); // AbortError se exceder o tempo
}

AbortError

Quando cancelado, todos os streams são fechados. O arquivo orders.csv.gz neste ponto está apenas parcialmente escrito. Em código real, você deve excluir esse arquivo no bloco catch, ou escrever em um arquivo temporário e fazer rename quando terminar.

7. Testando Transform

A lógica mais propensa a erros é a emenda de linhas cortadas entre dois chunks. Testamos usando o test runner nativo node:test. A fonte de dados fictícia é Readable.from com chunks propositalmente cortados no meio de uma linha e no meio de 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('emendar linhas 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 linha sem \n
  ]);

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

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

test('não corrompe caracteres UTF-8 multibyte', 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 no meio da letra "ễ"

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

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

Executar os testes:

node --test

✔ emendar linhas cortadas entre chunks (3.1ms)
✔ não corrompe caracteres UTF-8 multibyte (0.6ms)
ℹ tests 2
ℹ pass 2
ℹ fail 0

Neste teste, usar .pipe() é aceitável porque Readable.from roda na memória e não há I/O que exija limpeza. Se você remover StringDecoder e usar chunk.toString(), o segundo teste falhará com uma string contendo �.

8. Erros comuns e quando não usar streams

Usar on('data') com handler async. Por exemplo stream.on('data', async chunk => await db.insert(chunk)). O event emitter não aguarda a Promise, então o stream ainda empurra dados em velocidade máxima e milhares de insert rodam em paralelo. Use for await ou um passo async generator em pipeline.

Esquecer de tratar erros. Um stream emitindo 'error' sem um listener fará o processo inteiro falhar. Com pipeline em forma de Promise, você só precisa de um try/catch em um lugar.

Assumir que um chunk é uma linha ou um registro. Chunks são cortados por byte. Sempre separe linhas usando readline ou um Transform com buffer tail como na seção 6.2.

Definir highWaterMark muito grande. Colocar 100 MB para “ler mais rápido” quase não ajuda, e ainda perde os benefícios de memória. Um valor de 64 KiB a 1 MiB geralmente é suficiente.

Acumular resultados em um array grande. Código como for await (...) rows.push(row) e depois processar volta ao problema exato de readFile. Processe ou escreva os dados assim que os receber.

Nota: Streams nem sempre são a melhor opção. Para arquivos de configuração de alguns KB, ou quando você precisa de todos os dados de uma vez (ordenar tudo, fazer parse de um arquivo JSON simples), readFile é mais simples e rápido. Use streams apenas quando o tamanho dos dados é grande, desconhecido antecipadamente ou vem da rede.

9. Conclusão

Você viu que readFile consome RAM várias vezes maior que o tamanho do arquivo. Streams mantêm a memória estável processando chunk a chunk. Os pontos principais:

• Use createReadStream com for await para ler por chunk e readline para ler por linha.
• Respeite backpressure: verifique o valor retornado por write(), ou deixe pipeline cuidar disso.
• Use pipeline do node:stream/promises em vez de .pipe() para que erros se propaguem pelos streams e recursos sejam limpos corretamente.
• Escreva Transform com manipulação de linhas incompletas e bytes UTF-8 incompletos, depois teste com chunks propositalmente cortados.

Com essas ferramentas nativas, você pode filtrar, transformar e comprimir arquivos de vários GB enquanto o processo consome apenas algumas dezenas de MB de RAM.