Lưu một CSV rất lớn vào mongoDB bằng mongoose

Chào mừng bạn đến với phát trực tuyến. Những gì bạn thực sự muốn là một "luồng sự kiện" xử lý đầu vào của bạn "từng đoạn một", và tất nhiên lý tưởng nhất là bằng một dấu phân cách chung, chẳng hạn như ký tự "dòng mới" mà bạn hiện đang sử dụng.

Để có nội dung thực sự hiệu quả, bạn có thể thêm cách sử dụng MongoDB "Bulk API" chèn để giúp bạn tải nhanh nhất có thể mà không chiếm hết bộ nhớ máy hoặc chu kỳ CPU.

Không ủng hộ vì có nhiều giải pháp khác nhau, nhưng đây là danh sách sử dụng dòng- gói luồng đầu vào để làm cho phần "line terminator" trở nên đơn giản.

Chỉ định nghĩa lược đồ theo "ví dụ":

var LineInputStream = require("line-input-stream"),
    fs = require("fs"),
    async = require("async"),
    mongoose = require("mongoose"),
    Schema = mongoose.Schema;

var entrySchema = new Schema({},{ strict: false })

var Entry = mongoose.model( "Schema", entrySchema );

var stream = LineInputStream(fs.createReadStream("data.txt",{ flags: "r" }));

stream.setDelimiter("\n");

mongoose.connection.on("open",function(err,conn) { 

    // lower level method, needs connection
    var bulk = Entry.collection.initializeOrderedBulkOp();
    var counter = 0;

    stream.on("error",function(err) {
        console.log(err); // or otherwise deal with it
    });

    stream.on("line",function(line) {

        async.series(
            [
                function(callback) {
                    var row = line.split(",");     // split the lines on delimiter
                    var obj = {};             
                    // other manipulation

                    bulk.insert(obj);  // Bulk is okay if you don't need schema
                                       // defaults. Or can just set them.

                    counter++;

                    if ( counter % 1000 == 0 ) {
                        stream.pause();
                        bulk.execute(function(err,result) {
                            if (err) callback(err);
                            // possibly do something with result
                            bulk = Entry.collection.initializeOrderedBulkOp();
                            stream.resume();
                            callback();
                        });
                    } else {
                        callback();
                    }
               }
           ],
           function (err) {
               // each iteration is done
           }
       );

    });

    stream.on("end",function() {

        if ( counter % 1000 != 0 )
            bulk.execute(function(err,result) {
                if (err) throw err;   // or something
                // maybe look at result
            });
    });

});

Vì vậy, nói chung giao diện "luồng" ở đó "chia nhỏ đầu vào" để xử lý "từng dòng một". Điều đó ngăn bạn tải mọi thứ cùng một lúc.

Các phần chính là "API hoạt động hàng loạt" từ MongoDB. Điều này cho phép bạn "xếp hàng" nhiều thao tác cùng một lúc trước khi thực sự gửi đến máy chủ. Vì vậy, trong trường hợp này với việc sử dụng "modulo", các lần ghi chỉ được gửi trên 1000 mục được xử lý. Bạn thực sự có thể làm bất cứ điều gì lên đến giới hạn 16MB BSON, nhưng hãy giữ cho nó có thể quản lý được.

Ngoài các hoạt động được xử lý hàng loạt, có một "giới hạn" bổ sung từ async > thư viện. Nó không thực sự bắt buộc, nhưng điều này đảm bảo rằng về cơ bản không vượt quá "giới hạn mô đun" của các tài liệu đang được xử lý bất kỳ lúc nào. Các "chèn" hàng loạt chung không tốn IO nào khác ngoài bộ nhớ, nhưng các lệnh gọi "thực thi" có nghĩa là IO đang xử lý. Vì vậy, chúng tôi chờ đợi hơn là xếp hàng đợi nhiều thứ hơn.

Chắc chắn có các giải pháp tốt hơn mà bạn có thể tìm thấy cho dữ liệu loại CSV "xử lý luồng" có vẻ như đây là giải pháp. Nhưng nói chung, điều này cung cấp cho bạn các khái niệm về cách thực hiện việc này theo cách hiệu quả về bộ nhớ mà không tốn các chu kỳ CPU.