来源文件
{
"_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,
}
实现这一目标的最佳和最佳的方法是什么?此外,如果这可以以某种方式组合成单个语句,其中还包括添加新集合中存在但未在源集合中的文档,这将是惊人的:-)
PS.我正在使用Pymongo,所以Pymongo的例子会更好,但我可以翻译一个mongo shell示例.
最佳答案 不确定原子更新是否可行.但是,您可以在一些混合操作中进行字符串处理,并以这样的方式解决这个问题,即迭代新集合和新集合中的每个文档:
>使用_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();
两种情况下的批量操作API都有助于通过在集合中的每1000个文档中仅发送一次请求来减少服务器上的IO负载.