Stream trong Node.js: xử lý file lớn mà không cạn bộ nhớ

1. Tổng quan

Đọc cả file vào bộ nhớ bằng fs.readFile rất tiện khi file nhỏ. Với file log vài GB hoặc file CSV hàng triệu dòng, cách này làm process phình RAM hoặc crash. Bài này hướng dẫn bạn xử lý file lớn bằng stream trong Node.js: đọc từng chunk, tách dòng, lọc dữ liệu, nén và ghi ra file mới. Bạn cũng sẽ hiểu backpressure và biết cách xử lý lỗi đúng với pipeline.

Mọi ví dụ dùng Node.js 20+ và ES module. File package.json cần có "type": "module". Bài không cần thư viện ngoài vì fs, stream, zlib và readline đều có sẵn.

2. Vấn đề với readFile

Trước tiên, ta tạo một file orders.csv khoảng 5 triệu dòng để thử nghiệm:

// 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'); // chờ buffer ghi bớt rồi mới ghi tiếp
}
out.end();

Đoạn này đã dùng backpressure. Mục 5 sẽ giải thích kỹ cơ chế đó. File tạo ra nặng khoảng 180 MB.

Giờ hãy đếm số đơn hàng FAILED theo cách “ngây thơ”:

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

Một file 180 MB đã tốn khoảng 1 GB RAM. Process phải giữ cùng lúc ba thứ: buffer gốc, chuỗi UTF-8 và mảng 5 triệu chuỗi con. Khi file lớn hơn khoảng 2 GB, readFile sẽ ném ERR_FS_FILE_TOO_LARGE và không đọc được nữa.

3. Bốn loại stream cơ bản

Stream là dữ liệu đi qua từng mảnh (chunk). Node.js có bốn loại stream:

Loại	Vai trò	Ví dụ
Readable	Nguồn phát dữ liệu	fs.createReadStream, process.stdin
Writable	Đích nhận dữ liệu	fs.createWriteStream, process.stdout
Duplex	Vừa đọc vừa ghi, hai chiều độc lập	TCP socket
Transform	Duplex có output phụ thuộc input	zlib.createGzip()

Mỗi stream có một buffer nội bộ. Giới hạn của buffer này do highWaterMark quyết định. Mặc định fs.createReadStream dùng 64 KiB. Stream ở object mode dùng 16 object.

4. Đọc file lớn theo chunk

4.1. Dùng for await trên Readable

Cách gọn nhất để tiêu thụ một Readable là 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

Mỗi chunk là một Buffer tối đa 64 KiB. Vòng lặp chỉ yêu cầu chunk tiếp theo sau khi xử lý xong chunk hiện tại. Vì vậy RAM gần như không phụ thuộc kích thước file.

4.2. Đọc từng dòng với readline

Chunk cắt theo byte, không theo dòng. Nếu cần xử lý từng dòng, hãy dùng readline:

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

const rl = readline.createInterface({
  input: createReadStream('orders.csv'),
  crlfDelay: Infinity, // coi \r\n là một lần xuống dòng
});

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

FAILED: 500000

Kết quả giống mục 2 nhưng RSS chỉ khoảng 50–60 MB. readline tự ghép phần dòng bị cắt giữa hai chunk, nên bạn không phải lo chuyện đó.

5. Ghi file và backpressure

Backpressure xảy ra khi nguồn phát dữ liệu nhanh hơn đích ghi. Ổ đĩa, network hay gzip đều có thể chậm hơn tốc độ đọc. Hàm writable.write() trả về false khi buffer nội bộ vượt highWaterMark. Đó là tín hiệu “dừng lại, chờ sự kiện drain”.

Ví dụ sau bỏ qua tín hiệu này:

// bad-write.mjs — KHÔNG làm thế này
import { createWriteStream } from 'node:fs';

const out = createWriteStream('big.txt');
for (let i = 0; i < 10_000_000; i++) {
  out.write(`line ${i}\n`); // bỏ qua giá trị trả về
}
out.end();

Vòng lặp chạy đồng bộ nên toàn bộ 10 triệu chuỗi bị dồn vào buffer của out trước khi ổ đĩa kịp ghi. RAM tăng vọt lên hàng GB. Cách sửa là kiểm tra giá trị trả về rồi await once(out, 'drain'), như file generate-orders.mjs ở mục 2.

Nếu bạn nối các stream với nhau bằng pipeline, backpressure được xử lý tự động. Đây là cách nên dùng trong đa số trường hợp.

6. Kết nối stream bằng pipeline

6.1. Sao chép và nén file

// 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 từ node:stream/promises trả về một Promise. Promise này resolve khi stream cuối ghi xong. Nếu bất kỳ stream nào lỗi, ví dụ file không tồn tại hoặc ổ đĩa đầy, Promise sẽ reject. Khi đó tất cả stream đều bị destroy, nên file descriptor không bị rò rỉ.

Lưu ý: Hạn chế dùng a.pipe(b).pipe(c). pipe không chuyển lỗi qua các stream. Nếu a lỗi, c vẫn mở mãi. Hãy dùng pipeline thay thế.

6.2. Viết Transform tách dòng

Giả sử bạn muốn lọc đơn FAILED rồi nén lại. Bạn cần một bước tách dòng có thể đặt vào pipeline. Ta tự viết một 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, // đầu ra là từng dòng (string), không phải Buffer

    transform(chunk, _encoding, callback) {
      const lines = (tail + decoder.write(chunk)).split('\n');
      tail = lines.pop(); // phần cuối có thể là dòng dở dang
      for (const line of lines) this.push(line);
      callback();
    },

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

Giải thích từng phần:

• readableObjectMode: true: phía đọc phát ra từng dòng như một object riêng. Phía ghi vẫn nhận Buffer.
• StringDecoder: chunk có thể cắt ngang giữa một ký tự UTF-8 nhiều byte, chẳng hạn chữ “ệ” trong “Nguyễn Văn Hiệp”. Gọi chunk.toString() trực tiếp sẽ sinh ký tự lỗi �. decoder.write() giữ lại các byte dở dang và ghép chúng vào lần gọi sau.
• tail: lưu phần dòng chưa kết thúc để ghép với chunk tiếp theo.
• flush: chạy khi input kết thúc. Bước này đẩy nốt dòng cuối nếu file không có \n ở cuối.

6.3. Lọc bằng async generator

pipeline chấp nhận async generator làm một bước. Nhờ vậy bạn có thể lọc mà không cần viết thêm class 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

Dữ liệu chảy qua năm bước. Mỗi bước chỉ giữ vài chunk trong bộ nhớ. Khi gzip chậm, generator bị treo tại yield và splitLines ngừng nhận chunk. Kết quả là createReadStream tạm dừng đọc đĩa. Đó chính là backpressure chạy tự động.

6.4. Hủy giữa chừng với AbortSignal

Với tác vụ dài, bạn nên cho phép hủy hoặc đặt 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) }, // hủy sau 500 ms
  );
} catch (err) {
  console.error(err.name); // AbortError nếu quá thời gian
}

AbortError

Khi bị hủy, mọi stream đều được đóng. File orders.csv.gz lúc này chỉ được ghi dở. Trong code thật, hãy xóa file đó trong khối catch, hoặc ghi ra file tạm rồi rename khi xong.

7. Kiểm thử Transform

Logic dễ sai nhất là việc ghép dòng bị cắt giữa hai chunk. Ta test bằng test runner có sẵn node:test. Nguồn dữ liệu giả là Readable.from với các chunk cố tình cắt ngang dòng và cắt ngang ký tự 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('ghép dòng bị cắt giữa các chunk', async () => {
  const source = Readable.from([
    Buffer.from('1,an,10,PAID\n2,bi'),
    Buffer.from('nh,20,FAILED\n3,chi,5,PAID'), // dòng cuối không có \n
  ]);

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

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

test('không làm hỏng ký tự UTF-8 nhiều byte', 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)]); // cắt giữa chữ "ễ"

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

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

Chạy test:

node --test

✔ ghép dòng bị cắt giữa các chunk (3.1ms)
✔ không làm hỏng ký tự UTF-8 nhiều byte (0.6ms)
ℹ tests 2
ℹ pass 2
ℹ fail 0

Ở bài test này, dùng .pipe() vẫn ổn vì Readable.from chạy trong bộ nhớ và không có I/O cần dọn dẹp. Nếu bạn bỏ StringDecoder và dùng chunk.toString(), test thứ hai sẽ fail với chuỗi chứa �.

8. Lỗi hay gặp và khi nào không nên dùng stream

Dùng on('data') với handler async. Ví dụ stream.on('data', async chunk => await db.insert(chunk)). Event emitter không chờ Promise, nên stream vẫn đẩy dữ liệu với tốc độ tối đa và hàng nghìn câu insert chạy song song. Hãy dùng for await hoặc một bước async generator trong pipeline.

Quên xử lý lỗi. Stream phát 'error' mà không có listener sẽ làm crash cả process. Với pipeline dạng Promise, bạn chỉ cần try/catch một chỗ.

Giả định chunk là một dòng hoặc một bản ghi. Chunk được cắt theo byte. Hãy luôn tách dòng bằng readline hoặc một Transform có buffer tail như mục 6.2.

Đặt highWaterMark quá lớn. Đặt 100 MB thì “để đọc nhanh hơn” gần như không giúp gì, mà còn làm mất lợi ích về bộ nhớ. Giá trị 64 KiB đến 1 MiB thường là đủ.

Gom kết quả vào một mảng lớn. Code kiểu for await (...) rows.push(row) rồi mới xử lý thì quay lại đúng vấn đề của readFile. Hãy xử lý hoặc ghi ra ngay khi nhận dữ liệu.

Lưu ý: Stream không phải lúc nào cũng là lựa chọn tốt. Với file cấu hình vài KB, hoặc khi bạn cần toàn bộ dữ liệu cùng lúc (sắp xếp toàn bộ, parse một file JSON đơn), readFile đơn giản và nhanh hơn. Chỉ dùng stream khi kích thước dữ liệu lớn, không biết trước, hoặc đến từ network.

9. Kết luận

Bạn đã thấy readFile tốn RAM gấp nhiều lần kích thước file. Stream giữ bộ nhớ ổn định nhờ xử lý từng chunk. Các điểm chính:

• Dùng createReadStream với for await để đọc theo chunk, và readline để đọc theo dòng.
• Tôn trọng backpressure: kiểm tra giá trị trả về của write(), hoặc để pipeline lo việc này.
• Dùng pipeline từ node:stream/promises thay cho .pipe() để lỗi được chuyển qua các stream và tài nguyên được dọn dẹp đúng.
• Viết Transform có xử lý dòng dở dang và byte UTF-8 dở dang, rồi test bằng các chunk cố tình cắt ngang.

Với các công cụ có sẵn này, bạn có thể lọc, chuyển đổi và nén file nhiều GB mà process chỉ tốn vài chục MB RAM.