Mục lục nội dung

Giới thiệu vấn đề

Khi xây dựng hệ thống Backend cần nhập dữ liệu hàng triệu bản ghi (ví dụ: log, dữ liệu cảm biến, hoặc file CSV lớn) vào cơ sở dữ liệu, việc đọc toàn bộ file vào bộ nhớ rồi thực hiện insertMany sẽ gây Out Of Memory và làm Server treo. Giải pháp hiệu quả là kết hợp Stream để đọc dữ liệu từng phần và Batch Insert để giảm số lần ghi vào DB.

Cơ chế Stream trong Node.js

Node.js cung cấp ba loại Stream chính: Readable, Writable và Transform. Khi làm việc với file lớn, chúng ta thường sử dụng fs.createReadStream để tạo một Readable Stream và đọc dữ liệu theo chunk (đoạn). Điều này cho phép Node.js duy trì event‑loop không bị block.

Ví dụ đọc CSV bằng Stream

const fs = require('fs');
const readline = require('readline');

const stream = fs.createReadStream('data/large-file.csv');
const rl = readline.createInterface({ input: stream });

rl.on('line', (line) => {
    // Xử lý mỗi dòng ở đây
    console.log('Dòng:', line);
});

rl.on('close', () => {
    console.log('Đọc file hoàn tất');
});

Trong ví dụ trên, readline tạo một Transform Stream để tách dữ liệu theo dòng, giúp chúng ta không cần tải toàn bộ file vào bộ nhớ.

Batch Insert vào MongoDB

MongoDB hỗ trợ phương thức insertMany cho phép chèn nhiều tài liệu trong một lần gọi. Khi kết hợp với Stream, chúng ta có thể gom n bản ghi thành một mảng và thực hiện chèn khi mảng đạt kích thước nhất định (ví dụ: 1000 bản ghi).

Triển khai Batch Insert

const { MongoClient } = require('mongodb');
const BATCH_SIZE = 1000;
let batch = [];

async function run() {
    const client = await MongoClient.connect('mongodb://localhost:27017');
    const db = client.db('mydb');
    const collection = db.collection('records');

    const stream = fs.createReadStream('data/large-file.csv');
    const rl = readline.createInterface({ input: stream });

    rl.on('line', async (line) => {
        const fields = line.split(',');
        const doc = { name: fields[0], age: Number(fields[1]), city: fields[2] };
        batch.push(doc);
        if (batch.length >= BATCH_SIZE) {
            rl.pause(); // tạm dừng đọc để chờ DB
            await collection.insertMany(batch);
            batch = [];
            rl.resume(); // tiếp tục đọc
        }
    });

    rl.on('close', async () => {
        if (batch.length > 0) {
            await collection.insertMany(batch);
        }
        console.log('Hoàn tất chèn dữ liệu');
        await client.close();
    });
}

run().catch(console.error);

Điểm đáng chú ý:

  • pause()/resume() giúp kiểm soát tốc độ đọc, tránh quá tải MongoDB.
  • Kích thước batch nên được điều chỉnh dựa trên RAM và throughput của DB.
  • Trong môi trường production, nên sử dụng ordered: false để giảm thời gian khi một vài bản ghi lỗi.

Tối ưu thêm với BulkWrite

MongoDB cung cấp bulkWrite cho phép thực hiện nhiều thao tác (insert, update, delete) trong một lệnh duy nhất. Khi dữ liệu cần thực hiện upsert hoặc update theo khóa, bulkWrite sẽ nhanh hơn insertMany vì giảm số lần round‑trip.

Ví dụ bulk upsert

const ops = batch.map(doc => ({
    updateOne: {
        filter: { email: doc.email },
        update: { $set: doc },
        upsert: true
    }
}));
await collection.bulkWrite(ops, { ordered: false });

Nhờ ordered: false, các thao tác không phụ thuộc lẫn nhau và MongoDB sẽ thực hiện song song, giảm thời gian tổng thể.

Kết luận

Sử dụng Stream để đọc dữ liệu lớn và kết hợp Batch Insert hoặc BulkWrite là cách tiếp cận chuẩn để xử lý hàng triệu bản ghi mà không làm treo Server. Khi áp dụng đúng kích thước batch, kiểm soát flow bằng pause()/resume(), và tận dụng các tính năng của MongoDB, bạn có thể đạt thời gian xử lý dưới vài phút cho dữ liệu hàng chục triệu bản ghi.

Để nắm vững toàn bộ quy trình từ Event Loop, Stream, tới kiến trúc Backend chuyên nghiệp, Tham khảo khóa học "Lập trình Back-End với NodeJS Express" tại đây.