はじめに

Node.jsで数ギガバイトにもなるような大容量ファイルを扱う際、最も一般的な課題はメモリ不足です。fs.readFile() のような便利なメソッドは、ファイル全体を一度にメモリに読み込もうとするため、ファイルサイズが大きくなると簡単にアプリケーションがクラッシュしてしまいます。

この問題を解決するのが「Streams（ストリーム）」です。ストリームは、データを一度にすべて読み込むのではなく、小さなチャンク（塊）に分割して少しずつ処理するための仕組みです。これにより、使用するメモリ量を一定に保ちながら、どれだけ大きなデータでも安定して処理できます。

この記事では、Node.js Streamsの基本的な概念から、実践的なファイル操作、データ加工までを具体的なコード例と共に解説していきます。ストリームを使いこなし、パフォーマンスの高いアプリケーション開発を目指しましょう。

Node.js Streamsとは？4つの基本タイプ

ストリームを理解する最も簡単な方法は、水の流れをイメージすることです。蛇口から出た水がパイプを通り、最終的にバケツに溜まるように、データも一連の流れとして扱うことができます。Node.jsには、このデータの流れを制御するための4つの基本的なストリームタイプが存在します。

  • Readable Streams (読み込み可能ストリーム): データを読み込むためのストリームです。ファイルやHTTPリクエストなどがデータソースとなります。蛇口に相当します。
  • Writable Streams (書き込み可能ストリーム): データを受け取って書き込むためのストリームです。ファイルやHTTPレスポンスなどが書き込み先となります。バケツに相当します。
  • Duplex Streams (デュプレックスストリーム): 読み込みと書き込みの両方が可能なストリームです。TCPソケットなどが代表例です。
  • Transform Streams (変換ストリーム): データを読み込み、加工・変換してから次のストリームに渡す特殊なDuplexストリームです。データの圧縮(zlib)や暗号化(crypto)などがこれにあたります。パイプの途中にある浄水器のようなものです。

これらのストリームを組み合わせることで、複雑なデータ処理パイプラインを効率的に構築できます。

実践！Readable Streamでファイルを読み込む

まずは最も基本的なReadable Streamを使って、ファイルの内容を少しずつ読み込んでみましょう。Node.jsのfsモジュールには、ストリームを生成するためのcreateReadStream()メソッドが用意されています。

以下のコードは、large-file.txtというファイルをチャンク単位で読み込み、コンソールに出力する例です。ファイル全体をメモリに保持するのではなく、小さなデータチャンクが利用可能になるたびに'data'イベントが発火します。すべてのデータを読み込み終えると'end'イベントが、エラーが発生した場合は'error'イベントが発火します。

const fs = require('fs');

const readableStream = fs.createReadStream('large-file.txt', { encoding: 'utf8' });

// データチャンクが読み込まれるたびに実行
readableStream.on('data', (chunk) => {
  console.log('--- New Chunk ---');
  console.log(chunk.substring(0, 100)); // チャンクの先頭100文字だけ表示
});

// ファイルの読み込みが完了したときに実行
readableStream.on('end', () => {
  console.log('Finished reading file.');
});

// エラーが発生したときに実行
readableStream.on('error', (error) => {
  console.error(`Error: ${error.message}`);
});

この方法なら、たとえファイルが10GBあっても、アプリケーションが使用するメモリはごくわずかで済みます。これがストリームの最大の利点です。

パイプ(pipe)でストリームを繋ぐ

ストリームの真価は、複数のストリームを連結することで発揮されます。そのための強力なメソッドが.pipe()です。readableStream.pipe(writableStream)のように記述することで、読み込みストリームからの出力を自動的に書き込みストリームの入力へと流し込むことができます。

以下の例では、input.txtを読み込むReadable Streamを作成し、その内容をoutput.txtに書き込むWritable Streamに.pipe()で接続しています。これにより、効率的なファイルコピーが実現できます。

const fs = require('fs');

const readableStream = fs.createReadStream('input.txt');
const writableStream = fs.createWriteStream('output.txt');

// 読み込みストリームを書き込みストリームにパイプする
readableStream.pipe(writableStream);

writableStream.on('finish', () => {
  console.log('File copied successfully.');
});

readableStream.on('error', (error) => {
  console.error(`Error reading file: ${error.message}`);
});

writableStream.on('error', (error) => {
  console.error(`Error writing file: ${error.message}`);
});

.pipe()メソッドは、データの流量を自動的に調整する「背圧(backpressure)」という仕組みも内蔵しています。書き込みが追いつかない場合、読み込み速度を自動的に落としてくれるため、開発者は複雑な流量制御を意識する必要がなく、非常にシンプルで信頼性の高いコードを書くことができます。

応用例: Transform StreamでCSVデータを加工する

Transform Streamを使えば、データの流れの途中で加工処理を挟むことができます。例えば、大容量のCSVファイルから特定の列だけを抽出する、といった処理に最適です。

streamモジュールのTransformクラスを継承し、_transform()メソッドを実装することで、独自の変換ストリームを作成できます。このメソッドは、上流のストリームからチャンクを受け取り、加工した結果をthis.push()で下流に流します。

以下の例では、CSVの各行をパースし、2列目のデータだけを抽出する簡単なTransform Streamを作成しています。これを読み込みストリームと書き込みストリームの間に.pipe()で挟むことで、データ加工パイプラインを構築します。

const fs = require('fs');
const { Transform } = require('stream');

// 特定の列を抽出するTransform Stream
const columnExtractor = new Transform({
  transform(chunk, encoding, callback) {
    const lines = chunk.toString().split('\n');
    for (const line of lines) {
      const columns = line.split(',');
      // 2列目(インデックス1)が存在すればプッシュする
      if (columns.length > 1) {
        this.push(columns[1] + '\n');
      }
    }
    callback();
  }
});

fs.createReadStream('large-data.csv')
  .pipe(columnExtractor)
  .pipe(fs.createWriteStream('extracted-column.txt'));

console.log('CSV processing started...');

注意: この例は簡略化のため、チャンクの境界でCSVの行が分断されるケースを考慮していません。実用的なCSVパーサーでは、そうした境界ケースを適切に処理する必要があります。

まとめ

この記事では、Node.js Streamsの基本から応用までを解説しました。ストリームを使えば、メモリ使用量を抑えながら大容量データを効率的に処理できることをご理解いただけたかと思います。

重要なポイントを振り返りましょう。

  • メモリ効率: ファイル全体をメモリに読み込まず、小さなチャンクで処理するため、メモリ消費を最小限に抑えられます。
  • コンポーザビリティ: .pipe()メソッドを使うことで、読み込み、変換、書き込みといった処理を柔軟に組み合わせ、宣言的なデータ処理パイプラインを構築できます。
  • パフォーマンス: 流量を自動制御する背圧の仕組みにより、安定したデータ転送が可能です。

ファイル処理だけでなく、HTTPリクエスト/レスポンス、データの圧縮・解凍(zlib)、暗号化(crypto)など、Node.jsの多くのコアモジュールがストリームベースで設計されています。ストリームをマスターすることは、より高度でパフォーマンスの高いNode.jsアプリケーションを開発するための重要なステップです。

記事一覧に戻る