产生一个 mongoinsert

Question

我的目标是插入非常大的 csv，所以我不会像这样使用 csv 流：

            var myCollection = db.collection(myCollectionId);

            var q = async.queue(Collection.insert.bind(myCollection), 10);

            csv()
            .from.path(myFilePath, {columns: true}) 
            .transform(function(data, index, cb){

                    q.push(data, function (err, res) {
                        if (err) return cb(err);
                        cb(null, res[0]);
                    });

            })
            .on('end', function () {

                q.drain = function() { 

                       //do some stufff
                };

            })
            .on('error', function (err) {
                res.end(500, err.message);
                console.log('on.error() executed');
            });

        });

但是当文件变得非常大，比如 70M+ 并且它正在流式传输它们时，我的服务器非常慢并且需要很长时间，并且当我尝试在网站上加载页面时，在此过程中它会昏昏欲睡。

为什么不能像这样使用 cron-job 执行 mongo 插入。我问是因为从 mongo 命令行执行相同的插入可能需要 30 秒。

附：不要介意 readFile 和 lines 部分，我这样做是因为我想测试在进程启动后何时将所有行插入到集合中（尚未实现）。

var cronJob = require('cron').CronJob;
var spawn = require('child_process').spawn; 
var fs = require('fs');
function MongoImportEdgeFile(dataID, filePath){

var scriptPath = "/local/main/db/mongodb-linux-x86_64-2.4.5/bin/mongoimport";
console.log("script path = "+scriptPath)
var output = "";

 fs.readFile(filePath, 'utf-8',function(err, data) {

        if (err){
            console.log(err)
            throw err;
        }

        //console.log('data = '+data);
        var lines = data.split('\n');
        console.log("total lines in file = " + lines);

        var job = new cronJob(new Date(), function() {
            // store reference to 'this', which is cronJob object.  needed to stop job after script is done executing.
            var context = this;

            // execute R script asynchronously
            var script = spawn(scriptPath, [" -d mydb -c Data_ForID_" + dataID + " --file " + filePath + " --type csv" ]);
            console.log("Executing R script via node-cron: " + scriptPath);

            // script has finished executing, so complete cron job and fire completion callback
            script.on('close', function() {
                console.log('inside script.on(close, function() for import');
                context.stop();
            });
        }, function() {
            // callback function that executes upon completion
            console.log("Finished executing import");

        }, true);

   });

}

Answer 1

您不应使用单独的insert 调用。你强迫mongo 对每个调用执行内部同步——我认为考虑到你的并行方法，情况更糟。

使用bulk insertion：就像使用array 调用insert() 一样简单。

Answer 2

您可以通过创建child process 直接从节点执行mongoimport。 Here's an article on using mongoimport to import a csv。你也可以json。

不知何故，我错过了在 cron 中使用 mongoimport 的部分。如果我理解正确，您似乎以某种方式知道要导入的 csv，并且您正在使用 cron 来检查它们。

您考虑过消息队列吗？这将允许您的处理器立即而不是间隔接收导入作业。这也会限制您的处理。

如果您需要更高的吞吐量，您可以创建附加到同一队列的附加侦听器进程。他们将竞争下一份工作。这将使您的解决方案能够扩展。