mongodb - 如果不存在，则将字段添加到 Mongo 中的文档中

Question

源文件

{ 
    "_id" : "12345", 
    "LastName" : "Smith", 
    "FirstName" : "Fred", 
    "ProfileCreated" : NumberLong(1447118831860), 
    "DropOut" : false, 
}

新建文档

{ 
    "_id" : "12345", 
    "LastName" : "Smith", 
    "FirstName" : "Fred", 
    "ProfileCreated" : NumberLong(1447118831860), 
    "DropOut" : true, 
    "LatestConsultation" : false,
}

我有两个集合，它们共享许多相同的文档 ID 和字段，但随着时间的推移，新文档将添加字段，或者将创建具有新 ID 的全新文档。

我想我知道如何使用 $setOnInsert 和 upsert = true 处理新文档，但我不确定如何最好地处理新字段的添加。对于在 _id 上与新字段匹配的两个集合中存在的文档，我要求的行为是将新字段添加到文档中，而不修改任何其他字段的值，即使它们已更改，如DropOut 值已更改的示例。我需要的结果文件是。

结果文档

{ 
    "_id" : "12345", 
    "LastName" : "Smith", 
    "FirstName" : "Fred", 
    "ProfileCreated" : NumberLong(1447118831860), 
    "DropOut" : false, 
    "LatestConsultation" : false,
}

---

实现此目标的最佳和最高效的方法是什么？此外，如果这可以以某种方式组合成一个语句，该语句还包括添加新集合中存在但源集合中不存在的文档，那将是惊人的:-)

附言。我正在使用 Pymongo，因此 Pymongo 示例会更好，但我可以翻译一个 mongo shell 示例。

Answer 1

不确定原子更新是否可行。但是，您可以串入一些混合操作并以迭代新集合和新集合中的每个文档的方式解决这个问题:

• 使用_id 字段查询旧集合。使用 findOne() 方法从旧集合返回与新集合的 _id 匹配的文档。
• 通过添加旧文档中不存在的新字段，用旧文档扩展新文档。
• 使用合并后的文档更新新集合。

---

以下基本的 mongo shell 示例演示了上述算法:

function merge(from, to) {
    var obj = {};
    if (!from) {
        from = {};
    } else {
        obj = from; 
    }
    for (var key in to) {
        if (!from.hasOwnProperty(key)) {
            obj[key] = to[key];
        }
    }
    return obj;
}

db.new_collection.find({}).snapshot().forEach(function(doc){
    var old_doc = db.old_collection.findOne({ "_id": doc._id }),
        merged_doc = merge(old_doc, doc);

    db.new_collection.update(
        { "_id": doc._id },
        { "$set": merged_doc }
    );
});

---

为了处理大型集合，使用批量 API 更好地利用您的更新，它提供更好的性能和通过以下方式完成的高效更新操作 批量发送更新请求，而不是针对每个请求发送每个更新操作(这很慢)。使用的方法是 bulkWrite() 函数，可以在上面的例子中应用为:

function merge(from, to) {
    var obj = {};
    if (!from) {
        from = {};
    } else {
        obj = from; 
    }
    for (var key in to) {
        if (!from.hasOwnProperty(key)) {
            obj[key] = to[key];
        }
    }
    return obj;
}

var ops = [];
db.new_collection.find({}).snapshot().forEach(function(doc){
    var old_doc = db.old_collection.findOne({ "_id": doc._id }),
        merged_doc = merge(old_doc, doc);

    ops.push({
        "updateOne": {
            "filter": { "_id": doc._id },
            "update": { "$set": merged_doc }
        }
    });

    if (ops.length === 1000) {
        db.new_collection.bulkWrite(ops);
        ops = [];
    }
});

if (ops.length > 0)  db.new_collection.bulkWrite(ops);

或者对于 MongoDB 2.6.x 和 3.0.x 版本使用此版本的 Bulk 操作:

var bulk = db.new_collection.initializeUnorderedBulkOp(),
    counter = 0;

db.new_collection.find({}).snapshot().forEach(function(doc){
    var old_doc = db.old_collection.findOne({ "_id": doc._id }),
        merged_doc = merge(old_doc, doc);

    bulk.find({ "_id": doc._id }).updateOne({ "$set": merged_doc });

    if (counter % 1000 === 0) {
        bulk.execute();
        bulk = db.new_collection.initializeUnorderedBulkOp();
    }
});

if (counter % 1000 !== 0 )  bulk.execute();

在这两种情况下，Bulk 操作 API 将通过在集合中每 1000 个文档中仅发送一次请求来帮助减少服务器上的 IO 负载以进行处理。