node.js - 在 NodeJS 中将 200'000 行以上的大型 csv 文件插入 MongoDB

Question

我正在尝试解析一个大的 csv 文件并将其插入 MongoDB，但是当文件扩展 100'000 行时，我得到了来自服务器的错误响应。而且我需要插入的文件通常在 200'000 行以上。

我已经尝试过批量插入 (insertMany) 和 Babyparse(Papaparse) 流式方法来逐行插入文件。但效果不佳。

Node 接口(interface):

router.post('/csv-upload/:id', multipartMiddleware, function(req, res) {

    // Post vartiables
    var fileId = req.params.id;
    var csv = req.files.files.path;

    // create a queue object with concurrency 5
    var q = async.queue(function(row, callback) {
        var entry = new Entry(row);
        entry.save();
        callback();
    }, 5);

    baby.parseFiles(csv, {
        header: true, // Includes header in JSON
        skipEmptyLines: true,
        fastMode: true,
        step: function(results, parser) {
            results.data[0].id = fileId;

            q.push(results.data[0], function (err) {
                if (err) {throw err};
            });
        },
        complete: function(results, file) {
            console.log("Parsing complete:", results, file);
            q.drain = function() {
                console.log('All items have been processed');
                res.send("Completed!");
            };
        }
    });
});

这种流式处理方法会导致:POST SERVER net::ERR_EMPTY_RESPONSE

不确定我是否正确使用了 async.queue。

有没有更好、更有效的方法来做到这一点，或者我做错了什么？

express 服务器:

// Dependencies
var express = require('express');
var path = require('path');
var bodyParser = require('body-parser');
var routes = require('./server/routes');
var mongoose = require("mongoose");
var babel = require("babel-core/register");
var compression = require('compression');
var PORT = process.env.PORT || 3000;
// Include the cluster module
var cluster = require('cluster');

mongoose.connect(process.env.MONGOLAB_URI || 'mongodb://localhost/routes');

  // Code to run if we're in the master process
 if (cluster.isMaster) {

    // Count the machine's CPUs
    var cpuCount = require('os').cpus().length;

    // Create a worker for each CPU
    for (var i = 0; i < cpuCount; i += 1) {
        cluster.fork();
    }

 // Code to run if we're in a worker process
 } else {
    // Express
    var app = express();

    app.use(bodyParser.json({limit: '50mb'}));
    app.use(bodyParser.urlencoded({limit: '50mb', extended: true}));

    // Compress responses
    app.use(compression());

    // Used for production build
    app.use(express.static(path.join(__dirname, 'public')));

    routes(app);

    // Routes
    app.use('/api', require('./server/routes/api'));

    app.all('/*', function(req, res) {
        res.sendFile(path.join(__dirname, 'public/index.html'));
    });

    // Start server
    app.listen(PORT, function() {
        console.log('Server ' + cluster.worker.id + ' running on ' + PORT);
    });
}

Answer 1

处理导入:

很好的问题，根据我的经验，将 csv 插入 mongo 的最快方法是通过命令行:

mongoimport -d db_name -c collection_name --type csv --file file.csv --headerline 

我不相信 mongoose 有调用 mongoimport 的方法(如果我错了，请有人纠正我)

但是直接通过node调用就很简单了:

var exec = require('child_process').exec;
var cmd = 'mongoimport -d db_name -c collection_name --type csv --file file.csv --headerline';

exec(cmd, function(error, stdout, stderr) {
  // do whatever you need during the callback
});

以上内容必须修改为动态的，但应该是不言自明的。

处理上传:

从前端客户端上传文件是另一个挑战。

如果您向服务器发出请求并且在 60 秒内没有得到响应(可能是您在上面提到的)，大多数浏览器都会超时

一种解决方案是打开一个套接字连接(在 npm 中搜索 socket.io)以获取详细信息。这将创建与服务器的持续连接，并且不受超时限制。

如果上传不是问题，并且超时是由于解析/插入速度慢，那么一旦您实现了上述操作，您就不必担心这一点。

其他注意事项:

我不确定您需要向用户发送回什么，或者需要进行什么解析。但这既可以在正常的请求/响应周期之外完成，也可以在一个请求/响应周期中需要时在套接字连接期间进行处理。