如何使用 NodeJS 在 MongoDB 中使用批处理插入数百万个文档答案

【问题标题】：How to insert millions of document using batch process in MongoDB with NodeJS如何使用 NodeJS 在 MongoDB 中使用批处理插入数百万个文档
【发布时间】：2019-05-18 07:41:03
【问题描述】：

我编写了 nodejs 脚本来读取 JSON 文件并将多条记录插入到 mongo 集合中，我有数百万条记录，我不想一次性插入所有文档。我想每秒插入 300 个文档并休眠 30 秒，然后再插入 300 个等等？我是 NodeJS 的新手——请问我如何用下面的代码实现这一点？感谢您的帮助和支持。

app.js

const mongoClient = require("mongodb").MongoClient;
const util = require('util');
const fs = require('fs');

let database = null;
new mongoClient('mongodb://localhost:3000/', {
    auth: {
        user: 'admin',
        password: 'password',
    }
}).connect(
    (err, db) => {
        if (err) return console.error(err);
        database = db.db('myDB');
        fs.readFile('data.json', 'utf8', function(err, data) {
            if (err) throw err;
            var json = JSON.parse(data);
            database.collection("test").insertMany(json, function(err, doc) {
                console.log("Documents inserting");
                if (err) throw err;
            });
            //db.close();
        });
    });

示例数据： - 我在一个文件中有数百万条这样的记录。

 [{
    "firstName": "Ariel",
    "lastName": "Bailey"
 }, {
    "firstName": "Lura",
    "lastName": "Buckridge"
 }, {
    "firstName": "Milton",
    "lastName": "Macejkovic"
 }, {
    "firstName": "Carolyn",
    "lastName": "Hegmann"
 }, {
    "firstName": "Sid",
    "lastName": "Beer"
 }]

【问题讨论】：

json 文件大小是多少？睡觉有什么意义？
@yeya JSON 文件大小为 40-45mb，睡眠点在每插入 300 条记录之后。
您发现了哪种错误？
@PyaePhyoeShein 我卡住了如何读取数百万条记录并插入一批 300 条记录并暂停一段时间并在 30 秒后再次插入。这个过程一直持续到所有记录都插入为止。我当前的代码是一次性插入所有数百万条记录 - 这是我不想要的。
@learngroovy 好的，我现在正在做。

标签： node.js mongodb

【解决方案1】：

您可以像这样使用 setTimeout 来“批量”插入：

fs.readFile('data.json', 'utf8', function (err, data) {
  if (err) throw err;
  var json = JSON.parse(data);
  processRecords(0, 300, 30 * 1000);

  function processRecords(startIdx, n, delay) {
    if (startIdx >= json.length) return db.close();

    database.collection("test").insertMany(json.slice(startIdx, startIdx + n), function (err, doc) {
      if (err) throw err;

      setTimeout(() => processRecords(startIdx + n, n, delay), delay);
    });
  }
});

【讨论】：

谢谢，它可以工作，但睡眠只发生在前 300 条记录中——30 秒后，其他记录会一次性插入.. :(
忘记在 setTimeout 调用中为 processRecords 提供第三个参数。立即尝试。

【解决方案2】：

这是您想要获得的基本概念。老实说，它不是 100% 完美的，剩下的就是你的努力。

var ids = 0, offset = 10000;

function readJson() {
    var json = /* read JSON file */;
    return json;
}

function splitWithBatch(ids, offset) {
    var jsonObj = {};
    for(var i = ids; i < offset; i++){
        jsonObj.push(json[i]);
    }
    return Q.resolve(jsonObj);
}

function callSending(ids) {
    return splitWithBatch(ids, 0).then(ProcessToSave);
}

function ProcessToSave(json) {
    var quantityLimit = 1000;
    return SendToMongo(json).then(doNextBatch);
    function doNextBatch() {
        if (json.length === quantityLimit) {
            return splitWithBatch(ids, offset + quantityLimit);
        } else {
            return Q.resolve(null);
        }
    }
}

function SendToMongo(json) {
    database.collection('test').insertMany(json, function(err, doc) {
        if (err) throw err;
    });
}

readJson().then(callSending).then(
    function(){
        console.log('done');
    },
    function (err){
        console.log('err', err.stack, err);
    }
);

【讨论】：