Selamat datang di streaming. Yang benar-benar Anda inginkan adalah "aliran peristiwa" yang memproses masukan Anda "satu per satu", dan tentu saja idealnya dengan pembatas umum seperti karakter "baris baru" yang sedang Anda gunakan.
Untuk hal-hal yang sangat efisien, Anda dapat menambahkan penggunaan MongoDB "Bulk API" sisipan untuk membuat pemuatan Anda secepat mungkin tanpa menghabiskan semua memori mesin atau siklus CPU.
Tidak menganjurkan karena ada berbagai solusi yang tersedia, tetapi berikut adalah daftar yang menggunakan line- paket input-stream untuk membuat bagian "terminator baris" menjadi sederhana.
Definisi skema hanya dengan "contoh":
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
});
});
});
Jadi umumnya antarmuka "aliran" di sana "memecah input" untuk memproses "satu baris pada satu waktu". Itu menghentikan Anda dari memuat semuanya sekaligus.
Bagian utamanya adalah "Bulk Operations API" dari MongoDB. Ini memungkinkan Anda untuk "mengantre" banyak operasi sekaligus sebelum benar-benar mengirim ke server. Jadi dalam hal ini dengan penggunaan "modulo", penulisan hanya dikirim per 1000 entri yang diproses. Anda benar-benar dapat melakukan apa saja hingga batas BSON 16 MB, tetapi tetap dapat dikelola.
Selain operasi yang diproses secara massal, ada "pembatas" tambahan dari async Perpustakaan. Ini tidak benar-benar diperlukan, tetapi ini memastikan bahwa pada dasarnya tidak lebih dari "batas modulo" dokumen sedang diproses setiap saat. Batch umum "sisipan" datang tanpa biaya IO selain memori, tetapi panggilan "eksekusi" berarti IO sedang diproses. Jadi kami menunggu daripada mengantri lebih banyak.
Pasti ada solusi yang lebih baik yang dapat Anda temukan untuk data jenis CSV "pemrosesan aliran" yang tampaknya seperti ini. Namun secara umum, ini memberi Anda konsep tentang cara melakukannya dengan cara yang efisien memori tanpa memakan siklus CPU juga.
