Node.jsのStream：メモリを圧迫しない大きなファイル処理

1. 概要

ファイル全体をfs.readFileでメモリに読み込むのは、ファイルが小さいときは便利です。しかし数GBのログファイルや数百万行のCSVファイルの場合、このアプローチはプロセスのRAMを急増させるか、クラッシュさせる可能性があります。この記事では、Node.jsのstreamを使って大きなファイルを処理する方法を紹介します：チャンクごとに読み込み、行を分割し、データをフィルタリング、圧縮して、新しいファイルに書き込みます。backpressureについて理解し、pipelineを使った正しいエラー処理方法も学べます。

すべての例はNode.js 20以上とESモジュールを使用しています。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でこの仕組みを詳しく説明します。作成されるファイルは約180MBです。

では、素朴な方法で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

180MBのファイルで約1GBのRAMが消費されました。プロセスは3つのものを同時に保持する必要があります：元のバッファ、UTF-8文字列、および500万の部分文字列の配列。ファイルが約2GBを超えると、readFileはERR_FS_FILE_TOO_LARGEをスローし、読み込めなくなります。

3. 4つの基本的なstreamタイプ

Streamはデータがチャンク（断片）ごとに流れるものです。Node.jsには4種類のstreamがあります：

種類	役割	例
Readable	データソース	fs.createReadStream、process.stdin
Writable	データの受け取り先	fs.createWriteStream、process.stdout
Duplex	読み取りと書き込みの両方、独立した双方向	TCPソケット
Transform	出力が入力に依存するDuplex	zlib.createGzip()

各streamには内部バッファがあります。このバッファの上限はhighWaterMarkで決まります。デフォルトではfs.createReadStreamは64KiBを使用します。object modeのstreamは16オブジェクトを使用します。

4. ファイルをチャンクごとに読み込む

4.1. Readableでfor awaitを使用する

Readableを処理する最も簡潔な方法は、非同期反復です：

// 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は最大64KiBの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を1回の改行として扱う
});

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

FAILED: 500000

結果はセクション2と同じですが、RSSはわずか50～60MBです。readlineは2つのチャンク間で切れた行の部分を自動的に結合するため、それについて心配する必要はありません。

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();

ループは同期的に実行されるため、10百万個の文字列すべてが、ディスクが書き込む前にoutのバッファに蓄積されます。RAMは数GBに急増します。修正方法は、戻り値をチェックしてからawait once(out, 'drain')を実行することです。セクション2のgenerate-orders.mjsファイルのように。

pipelineを使ってstreamを接続すると、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します。いずれかのstreamでエラーが発生した場合、例えばファイルが存在しないか、ディスクがいっぱいの場合、Promiseは拒否されます。その場合、すべての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, // 出力は各行（文字列）であり、Bufferではありません

    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：チャンクは「Nguyễn Văn Hiệp」の「ệ」など、複数バイトのUTF-8文字の途中で切られる可能性があります。chunk.toString()を直接呼び出すと、エラー文字�が生成されます。decoder.write()は不完全なバイトを保持し、次の呼び出しで結合します。
• tail：次のチャンクと結合するために、未完了の行の部分を保存します。
• flush：入力が終了したときに実行されます。ファイルの最後に\nがない場合、このステップは最後の行をプッシュします。

6.3. 非同期ジェネレータを使ってフィルタリングする

pipelineは非同期ジェネレータを1つのステップとして受け入れます。これにより、追加の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が遅い場合、ジェネレータは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) }, // 500ms後にキャンセル
  );
} catch (err) {
  console.error(err.name); // タイムアウトの場合はAbortError
}

AbortError

キャンセル時に、すべてのstreamが閉じられます。ファイルorders.csv.gzは部分的にのみ書き込まれています。実際のコードでは、catchブロック内でそのファイルを削除するか、一時ファイルに書き込んでから完了時にrenameしてください。

7. Transformのテスト

最も間違いやすいロジックは、2つのチャンク間で切れた行の結合です。組み込みのテストランナー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('2つのチャンク間で切れた行を結合する', 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

✔ 2つのチャンク間で切れた行を結合する (3.1ms)
✔ 複数バイトのUTF-8文字を破損させない (0.6ms)
ℹ tests 2
ℹ pass 2
ℹ fail 0

このテストでは、Readable.fromがメモリ内で実行され、クリーンアップが必要なI/Oがないため、.pipe()を使用しても問題ありません。StringDecoderを削除してchunk.toString()を使用すると、2番目のテストは�を含む文字列で失敗します。

8. よくあるエラーとstreamを使わない方が良い場合

on('data')で非同期ハンドラーを使用する。 例：stream.on('data', async chunk => await db.insert(chunk))。イベントエミッターはPromiseを待たないため、streamは引き続き最大速度でデータをプッシュし、数千のinsertが並行して実行されます。for awaitまたはpipeline内の非同期ジェネレータステップを使用してください。

エラー処理を忘れる。 リスナーのない'error'イベントはプロセス全体をクラッシュさせます。Promise形式のpipelineでは、1か所でtry/catchするだけで済みます。

チャンクが1行またはレコードであると想定する。 チャンクはバイト単位で切られます。readlineまたはセクション6.2のようなtailバッファを持つTransformを使用して、常に行を分割してください。

highWaterMarkを大きすぎるに設定する。 100MBに設定しても「より高速に読むため」は効果がなく、むしろメモリの利点が失われます。64KiBから1MiBの値で通常は十分です。

結果を大きな配列に集約する。 その後で処理するfor await (...) rows.push(row)の種類のコードは、readFileの問題に戻ります。データを受け取ったときに、すぐに処理または書き込んでください。

注意： Streamが常に良い選択とは限りません。数KBの設定ファイルの場合、またはすべてのデータを同時に必要とする場合（全体をソートする、単一のJSONファイルを解析する）は、readFileがより簡単で高速です。データサイズが大きい、事前に不明、またはネットワークから来ている場合にのみstreamを使用してください。

9. 結論

あなたはreadFileがファイルサイズの何倍もRAMを消費することを見ました。Streamはチャンクごとの処理によってメモリを安定に保ちます。主要なポイント：

• createReadStreamをfor awaitで使ってチャンク単位で読み込み、readlineで行単位で読み込みます。
• backpressureを尊重してください：write()の戻り値をチェックするか、pipelineに任せてください。
• エラーがstream間で転送され、リソースが適切にクリーンアップされるように、.pipe()ではなくnode:stream/promisesのpipelineを使用してください。
• 不完全な行と不完全なUTF-8バイトを処理するTransformを書いて、意図的にチャンクを横切るテストを実行してください。

これらの組み込みツールを使用すると、プロセスはわずか数十MBのRAMで数GBのファイルをフィルタリング、変換、圧縮できます。