Mongodb检查重复记录以进行批量插入

Question

我的收藏有以下记录：

{ "_id":"1", "field1":"foo","field2":"xyz", "field3":"something" ...}
{ "_id":"2", "field1":"bar","field2":"xyz", "field3":"something" ...}
{ "_id":"3", "field1":"foo","field2":"abc", "field3":"something" ...}
{ "_id":"4", "field1":"bar","field2":"lmn", "field3":"something" ...}

在插入新记录之前，我需要检查是否已存在具有相同field1和field2值的记录。然后，如果该请求已经存在，则将其丢弃。如果我一次插入一个记录，我可以做到这一点。如果我进行批量插入（即在我插入文档数组时），该如何处理？

我有[field1, field2]需要查询EX 的组合数组：

queryArray=[ { "field1":"foo","field2":"xyz"},
             { "field1":"bar","field2":"lmn"} ]

预期结果：

result=[  { "_id":"1", "field1":"foo","field2":"xyz", "field3":"something" ...},
          { "_id":"4", "field1":"bar","field2":"lmn", "field3":"something" ...}]

Answer 1

在两个字段上创建唯一的复合索引

db.collection.createIndex( { "field1": 1, "field2": 1 }, { "unique": true } )

使用该insertMany()方法进行批量插入，但将ordered选项设置为false，因为这将确保尝试所有写操作，即使有错误也是如此。错误后，有序操作将停止，而无序操作将继续处理队列中所有剩余的写操作：

var queryArray = [ 
    { "field1": "foo", "field2": "xyz" },
    { "field1": "bar", "field2": "lmn" }
];
try { db.collection.insertMany(queryArray, { "ordered": false }); } 
catch (e) {  print (e); }

这将输出一个文档

{
    "acknowledged" : true,
    "insertedIds" : [ 
        ObjectId("57443e6fa58e5654f3a6c5ae"), 
        ObjectId("57443e6fa58e5654f3a6c5af")
    ]
}

如果操作运行时具有写关注点，则结果文档将显示确认为true的字段；如果禁用写关注点，则结果将显示为false的字段；_id对于每个成功插入的文档，将显示一个数组。

由于queryArray中的文档不包含_id，因此mongod为每个文档创建并添加_id字段，并为其分配唯一的ObjectId值。既然你在两个领域实行独特field1和field2，上面显示的尝试写入的操作从而无序它继续处理任何剩余的写操作。

假设您删除了有序选项（默认情况下将其设置为true），那么您将从操作中获得以下输出：

var queryArray = [ 
    { "field1": "foo", "field2": "xyz" },
    { "field1": "bar", "field2": "lmn" }
];
try { db.collection.insertMany(queryArray); } 
catch (e) {  print (e); }

控制台输出：

{
    "name" : "BulkWriteError",
    "message" : "write error at item 0 in bulk operation",
    "ok" : undefined,
    "nInserted" : 0,
    "nUpserted" : 0,
    "nMatched" : 0,
    "nModified" : 0,
    "nRemoved" : 0,
    "getUpsertedIds" : function () {
      return bulkResult.upserted;
    },
    "getUpsertedIdAt" : function (index) {
      return bulkResult.upserted[index];
    },
    "getRawResponse" : function () {
      return bulkResult;
    },
    "hasWriteErrors" : function () {
      return bulkResult.writeErrors.length > 0;
    },
    "getWriteErrorCount" : function () {
      return bulkResult.writeErrors.length;
    },
    "getWriteErrorAt" : function (index) {
      if(index < bulkResult.writeErrors.length) {
        return bulkResult.writeErrors[index];
      }
      return null;
    },
    "getWriteErrors" : function () {
      return bulkResult.writeErrors;
    },
    "hasWriteConcernError" : function () {
      return bulkResult.writeConcernErrors.length > 0;
    },
    "getWriteConcernError" : function () {
      if(bulkResult.writeConcernErrors.length == 0) {
        return null;
      } else if(bulkResult.writeConcernErrors.length == 1) {
        // Return the error
        return bulkResult.writeConcernErrors[0];
      } else {

        // Combine the errors
        var errmsg = "";
        for(var i = 0; i < bulkResult.writeConcernErrors.length; i++) {
          var err = bulkResult.writeConcernErrors[i];
          errmsg = errmsg + err.errmsg;
          // TODO: Something better
          if (i != bulkResult.writeConcernErrors.length - 1) {
            errmsg = errmsg + " and ";
          }
        }

        return new WriteConcernError({ errmsg : errmsg, code : WRITE_CONCERN_FAILED });
      }
    },
    "tojson" : function (indent, nolint) {
      return tojson(bulkResult, indent, nolint);
    },
    "toString" : function () {
      return "BulkWriteError(" + this.tojson() + ")";
    },
    "shellPrint" : function () {
      return this.toString();
    },
    "hasErrors" : function () {
      return this.hasWriteErrors() || this.hasWriteConcernError();
    },
    "toSingleResult" : function () {
      if(singleBatchType == null) throw Error(
          "Cannot output single WriteResult from multiple batch result");
      return new WriteResult(bulkResult, singleBatchType, writeConcern);
    },
    "stack" : "BulkWriteError({\n\t\"writeErrors\" : [\n\t\t{\n\t\t\t\"index\" : 0,\n\t\t\t\"code\" : 11000,\n\t\t\t\"errmsg\" : \"E11000 duplicate key error index: test.collection.$field1_1_field2_1 dup key: { : \\\"foo\\\", : \\\"xyz\\\" }\",\n\t\t\t\"op\" : {\n\t\t\t\t\"_id\" : ObjectId(\"574441aea58e5654f3a6c5b6\"),\n\t\t\t\t\"field1\" : \"foo\",\n\t\t\t\t\"field2\" : \"xyz\"\n\t\t\t}\n\t\t}\n\t],\n\t\"writeConcernErrors\" : [ ],\n\t\"nInserted\" : 0,\n\t\"nUpserted\" : 0,\n\t\"nMatched\" : 0,\n\t\"nModified\" : 0,\n\t\"nRemoved\" : 0,\n\t\"upserted\" : [ ]\n})\nBulkWriteError@src/mongo/shell/bulk_api.js:372:44\nBulkWriteResult/this.toError@src/mongo/shell/bulk_api.js:335:16\nBulk/this.execute@src/mongo/shell/bulk_api.js:1162:1\nDBCollection.prototype.insertMany@src/mongo/shell/crud_api.js:279:5\n@(shell):1:7\n",
    "toResult" : function () {
      return new BulkWriteResult(bulkResult, singleBatchType, writeConcern);
    }
}

着重于返回的写错误：

"E11000 duplicate key error index: test.collection.$field1_1_field2_1 dup key: { : \\\"foo\\\", : \\\"xyz\\\" }\"

除了insertMany()方法之外，您还可以尝试Bulk()API方法，特别是initializeUnorderedBulkOp()在创建唯一的复合索引之后，需要调用该方法以进行无序的批量插入。

考虑上述情况的以下示例：

db.collection('collectionName', function(err, collection) {
    var bulk = collection.initializeUnorderedBulkOp();
    counter = 0;

    queryArray.forEach(function (doc){
        bulk.insert(doc);
        counter++;

        if (counter % 1000 == 0) {
            bulk.execute(function(err, result) {
                // you could do something with results, check for duplicate errors
                bulk = collection.initializeUnorderedBulkOp(); // re-initialise
            }); 
        }
    });

    // Clean-up remaining operations in the queue 
    if (counter % 1000 != 0 ) {     
        bulk.execute(function(err, result) {
            // you could do something with results, check for duplicate errors
            console.log(result);
        });
    }
});