Node.js의 Stream: 메모리를 낭비하지 않으면서 큰 파일 처리하기

1. 개요

fs.readFile을 사용해서 전체 파일을 메모리에 로드하는 것은 파일이 작을 때는 편하다. 하지만 파일 크기가 몇 GB인 로그 파일이나 수백만 줄의 CSV 파일이면 프로세스의 RAM이 부풀어지거나 크래시한다. 이 글은 Node.js의 stream을 사용해서 큰 파일을 처리하는 방법을 설명한다: 한 번에 한 청크씩 읽고, 줄을 분리하고, 데이터를 필터링하고, 압축해서 새 파일에 쓴다. 또한 backpressure를 이해하고 pipeline으로 올바르게 에러를 처리하는 방법도 배우게 될 것이다.

모든 예제는 Node.js 20 이상과 ES module을 사용한다. package.json에는 "type": "module"이 필요하다. fs, stream, zlib, readline은 모두 내장되어 있으므로 외부 라이브러리가 필요 없다.

2. readFile의 문제점

먼저 테스트용으로 약 500만 줄의 orders.csv 파일을 만들어보자:

// 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'); // 버퍼가 비워질 때까지 기다린 다음 계속 쓰기
}
out.end();

이 코드는 이미 backpressure를 사용하고 있다. 섹션 5에서 이 메커니즘을 자세히 설명할 것이다. 생성된 파일 크기는 약 180 MB이다.

이제 “naive한” 방식으로 FAILED 주문의 개수를 세어보자:

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

180 MB 크기의 파일이 약 1 GB의 RAM을 사용했다. 프로세스는 동시에 세 가지를 메모리에 보관해야 한다: 원본 버퍼, UTF-8 문자열, 그리고 500만 개의 문자열 배열. 파일 크기가 약 2 GB 이상으로 커지면 readFile은 ERR_FS_FILE_TOO_LARGE 에러를 던지고 파일을 읽을 수 없다.

3. 네 가지 기본 stream 타입

Stream은 청크 단위로 데이터가 흐르는 것이다. Node.js에는 네 가지 stream 타입이 있다:

타입	역할	예시
Readable	데이터 소스	fs.createReadStream, process.stdin
Writable	데이터 수신처	fs.createWriteStream, process.stdout
Duplex	양쪽 모두 가능, 독립적인 양방향	TCP socket
Transform	Duplex인데 출력이 입력에 따라 달라짐	zlib.createGzip()

각 stream은 내부 버퍼를 가지고 있다. 이 버퍼의 크기는 highWaterMark로 결정된다. 기본적으로 fs.createReadStream은 64 KiB를 사용한다. 객체 모드의 stream은 16개의 객체를 사용한다.

4. 큰 파일을 청크 단위로 읽기

4.1. Readable에서 for await 사용하기

Readable을 소비하는 가장 간단한 방법은 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

각 chunk는 최대 64 KiB의 Buffer이다. 루프는 현재 청크를 처리한 후에만 다음 청크를 요청한다. 그래서 RAM은 파일 크기에 거의 영향을 받지 않는다.

4.2. readline으로 줄 단위로 읽기

청크는 바이트 기준으로 자르기 때문에 줄 기준이 아니다. 각 줄을 처리해야 한다면 readline을 사용하자:

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

const rl = readline.createInterface({
  input: createReadStream('orders.csv'),
  crlfDelay: Infinity, // \r\n을 한 번의 줄바꿈으로 간주
});

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

FAILED: 500000

결과는 섹션 2와 같지만 RSS는 약 50–60 MB만 사용한다. readline은 두 청크 사이에 끊긴 줄을 자동으로 붙여주므로 직접 처리할 필요가 없다.

5. 파일 쓰기와 backpressure

Backpressure는 데이터 소스의 속도가 수신처의 속도보다 빠를 때 발생한다. 디스크, 네트워크, gzip 모두 읽는 속도보다 느릴 수 있다. writable.write() 함수는 내부 버퍼가 highWaterMark를 초과하면 false를 반환한다. 이것은 “멈춰, drain 이벤트를 기다려”라는 신호이다.

다음 예제는 이 신호를 무시한다:

// bad-write.mjs — 이렇게 하지 마세요
import { createWriteStream } from 'node:fs';

const out = createWriteStream('big.txt');
for (let i = 0; i < 10_000_000; i++) {
  out.write(`line ${i}\n`); // 반환값 무시
}
out.end();

루프는 동기적으로 실행되므로 1천만 개의 문자열이 모두 out의 버퍼에 들어가고 나서야 디스크 쓰기가 시작된다. RAM이 수 GB까지 급증한다. 해결 방법은 반환값을 확인한 후 await once(out, 'drain')을 사용하는 것인데, 이는 섹션 2의 generate-orders.mjs 파일에서 볼 수 있다.

Stream을 pipeline으로 연결하면 backpressure가 자동으로 처리된다. 대부분의 경우 이것이 권장하는 방법이다.

6. pipeline으로 stream 연결하기

6.1. 파일 복사 및 압축

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

node:stream/promises의 pipeline은 Promise를 반환한다. 이 Promise는 마지막 stream의 쓰기가 완료되면 resolve된다. 파일이 없거나 디스크가 가득 차면 pipeline이 실패한다. 이 경우 모든 stream이 destroy되므로 파일 디스크립터가 누수되지 않는다.

주의: a.pipe(b).pipe(c) 사용을 피하자. pipe는 stream 간에 에러를 전달하지 않는다. a가 에러를 발생시켜도 c는 계속 열려있다. 대신 pipeline을 사용하자.

6.2. 줄 분리 Transform 작성

FAILED 주문을 필터링한 다음 압축한다고 가정하자. pipeline에 넣을 수 있는 줄 분리 단계가 필요하다. 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, // 출력은 각 줄(문자열)로 이루어진 objects

    transform(chunk, _encoding, callback) {
      const lines = (tail + decoder.write(chunk)).split('\n');
      tail = lines.pop(); // 마지막 부분은 불완전한 줄일 수 있음
      for (const line of lines) this.push(line);
      callback();
    },

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

각 부분 설명:

• readableObjectMode: true: 읽기 쪽은 각 줄을 별도의 객체로 전송한다. 쓰기 쪽은 여전히 Buffer를 받는다.
• StringDecoder: 청크가 여러 바이트로 이루어진 UTF-8 문자 중간에서 끊길 수 있다. 예를 들어 “Nguyễn Văn Hiệp”의 “ễ” 같은 경우. chunk.toString()을 직접 사용하면 잘못된 문자 �가 생긴다. decoder.write()은 불완전한 바이트를 보관했다가 다음 호출과 합친다.
• tail: 미완성 줄을 저장해서 다음 청크와 합친다.
• flush: 입력이 끝나면 실행된다. 파일 끝에 \n이 없는 경우 마지막 줄을 전송한다.

6.3. async generator로 필터링

pipeline은 async generator를 단계로 받아들인다. 덕분에 추가 Transform 클래스를 작성하지 않고 필터링할 수 있다:

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

데이터는 5개 단계를 거친다. 각 단계는 버퍼에 몇 개의 청크만 보관한다. gzip이 느리면 generator가 yield에서 멈춘다. splitLines가 청크 수신을 멈춘다. 그 결과 createReadStream이 디스크 읽기를 일시 중지한다. 이것이 자동 backpressure다.

6.4. AbortSignal로 중단하기

오래 걸리는 작업의 경우 취소 또는 타임아웃을 허용해야 한다:

// 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) }, // 500 ms 후 취소
  );
} catch (err) {
  console.error(err.name); // 시간 초과 시 AbortError
}

AbortError

취소되면 모든 stream이 닫힌다. 이때 orders.csv.gz 파일은 부분적으로만 작성된다. 실제 코드에서는 catch 블록에서 파일을 삭제하거나, 임시 파일에 쓴 후 완료 시 rename하는 것이 좋다.

7. Transform 테스트

가장 실수하기 쉬운 부분은 두 청크 사이에 끊긴 줄 처리와 UTF-8 바이트 처리다. node:test를 사용해서 테스트해보자. 테스트 데이터는 Readable.from으로 의도적으로 줄과 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('두 청크 사이에 끊긴 줄 붙이기', async () => {
  const source = Readable.from([
    Buffer.from('1,an,10,PAID\n2,bi'),
    Buffer.from('nh,20,FAILED\n3,chi,5,PAID'), // 마지막 줄에 \n 없음
  ]);

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

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

test('여러 바이트 UTF-8 문자 손상 방지', 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)]); // "ễ" 문자 중간에서 끊기

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

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

테스트 실행:

node --test

✔ 두 청크 사이에 끊긴 줄 붙이기 (3.1ms)
✔ 여러 바이트 UTF-8 문자 손상 방지 (0.6ms)
ℹ tests 2
ℹ pass 2
ℹ fail 0

이 테스트에서는 .pipe()를 사용해도 괜찮다. Readable.from은 메모리에서 실행되고 정리해야 할 I/O가 없기 때문이다. StringDecoder를 빼고 chunk.toString()을 직접 사용하면 두 번째 테스트는 � 문자가 포함된 문자열로 실패할 것이다.

8. 흔한 실수와 stream을 사용하지 말아야 할 때

async handler를 on('data')에 등록. 예: stream.on('data', async chunk => await db.insert(chunk)). Event emitter는 Promise를 기다리지 않으므로 stream이 최대 속도로 데이터를 계속 보낸다. 수천 개의 insert가 동시에 실행된다. 대신 for await을 사용하거나 pipeline에 async generator 단계를 추가하자.

에러 처리 누락. Stream이 listener 없이 'error'를 발생시키면 전체 프로세스가 크래시한다. Promise 형태의 pipeline을 사용하면 한 곳의 try/catch로 충분하다.

청크가 한 줄 또는 한 레코드라고 가정. 청크는 바이트 기준으로 자른다. 항상 readline 또는 섹션 6.2처럼 tail 버퍼가 있는 Transform으로 줄을 분리하자.

highWaterMark를 지나치게 크게 설정. 100 MB로 설정해서 “더 빠르게 읽기”를 기대하는 것은 거의 도움이 안 되면서 메모리 이점을 잃어버린다. 64 KiB에서 1 MiB 사이가 보통 충분하다.

결과를 큰 배열로 모으기. for await (...) rows.push(row) 후 처리하는 코드는 readFile의 문제로 돌아간다. 데이터를 받으면 바로 처리하거나 쓰자.

주의: Stream이 항상 좋은 선택은 아니다. 몇 KB의 설정 파일이거나 전체 데이터가 필요할 때 (전체 정렬, 단일 JSON 파일 파싱) readFile이 더 간단하고 빠르다. 데이터 크기가 크거나 미리 알 수 없거나 네트워크에서 오는 경우에만 stream을 사용하자.

9. 결론

readFile이 파일 크기의 몇 배 RAM을 사용한다는 것을 봤다. Stream은 각 청크를 처리하므로 메모리를 안정적으로 유지한다. 핵심 요점들:

• createReadStream과 for await로 청크를 읽고, readline으로 줄을 읽는다.
• Backpressure를 존중하자: write()의 반환값을 확인하거나 pipeline에 맡긴다.
• .pipe()보다 node:stream/promises의 pipeline을 사용하면 에러가 stream을 거쳐 전달되고 리소스가 올바르게 정리된다.
• 끊긴 줄과 끊긴 UTF-8 바이트를 처리하는 Transform을 작성한 다음 의도적으로 끊기는 청크로 테스트하자.

이러한 내장 도구들로 프로세스는 수십 MB의 RAM만 사용하면서 수십 GB의 파일을 필터링, 변환, 압축할 수 있다.