在 MongoDB 中合并文档

Question

我有一个庞大的 MongoDB 集合（约 50 万份文档）。

结构是这样的：

{'_id': '.....',
'passid':'ag325gdtew',
'text': '.......',
'count': '.......',
'title': '......',
'body': '.......'
}

passid 字段在许多文档中是相同的，我想将它们与每个字段的不同方式结合起来。

我想：

• 保持不变passid
• 在每个文档中加入文本和标题（文本+标题），然后在新文档的一个字段中加入最终字符串（文本1+文本2+文本3）
• 为每个计数创建一个字段列表 [count1, count2, count3]
• 删除正文字段

所以新的文档会是这样的：

{'_id': '.....',
'passid':'ag325gdtew',
'text': '.......', (string)
'count': ['..','...','..'] (list)
}

目前，我正在使用 Python 执行此操作，但文件很大，脚本会持续运行数小时。

我做了什么：

• 与组聚合以检索具有唯一passid 的列表
• 在该列表中迭代每个passid
• 使用 find(passid) 检索具有相同passid 的所有文档的游标
• 用python对字符串和列表进行连接和追加
• 删除旧文档
• 保存新的

正如我所说，这真的很耗时。你知道有什么更快的方法吗？

代码如下：

passids= db.collection.aggregate({ "$group": {"_id": '$passid'}})

for i in passids['result']:
    doc = {}
    doc['passid'] = i['_id']
    documents = db.collection.find({"passid": i['_id']})
    doc['count'] = []
    doc['text'] = ""

    for d in documents:
        doc['text'] = doc['text'] + " " + d['text']
        doc['text'] = doc['text'] + " " + d['title']
        doc['count'].append(d['count'])
        db.collection.remove(d)
    db.collection.save(doc)

Answer 1

如果您试图避免应用程序和数据库之间的大量网络流量，您的最佳选择通常是尝试在尽可能靠近数据库的位置（在网络方面）运行代码以获得最佳速度。

如果这是不可能的，并且真的只应该在“一次性”操作中使用，您可以使用db.eval()在服务器上运行代码

警告在考虑使用之前，您必须仔细阅读db.eval() 的手册页。虽然完成工作的最快方法有一些主要缺点需要考虑：

1. 这将在整个执行期间获得数据库上的写锁。
2. 除了获得写入锁之外，由于 JavaScript 实现的单线程特性，使用 JavaScript 解释器的“mapReduce”作业等其他任务将无法运行。
3. 这将无法在分片集群上运行，如果您的主机使用身份验证，则用户帐户将需要基本读写之外的特殊权限才能执行任务。

一旦考虑了以上所有因素，您就可以摆脱困境，承认方法存在并继续前进。

只要您可以处理输出到不同的集合，您就可以从 mapReduce 开始，这样可以简化逻辑

你可以定义一个映射器：

var mapper = function() {
   
   var passid = this.passid;
   delete this["_id"];
   delete this["body"];

   emit( passid, this );

};

然后定义一个reducer：

var reducer = function(key,values) {

    var reducedObject = {
        "text": "",
        "count": []
    };

    values.forEach(function(value) {
        reducedObject.text = reducedObject.text + " " + value.text;
        reducedObject.text = reducedObject.text + " " + value.title;
        reducedObject.push( value.count );
    });

    return reducedObject;

};

然后就可以运行mapReduce操作了：

db.collection.mapReduce(
    mapper,
    reducer,
    {
        "out": { "replace": "newcollection" }
    }
)

由于 mapReduce 输出就是这样，您不希望在最终输出中使用它，因此您可以像这样进行更改：

db.eval(function() {
    db.newcollection.find().forEach(function(doc) {
        var newDoc = {};
        for ( var k in doc.values ) {
            newDoc[k] = doc.values[k];
        }
        db.newcollection.update({ _id: doc._id }, newDoc );
    });
})

这会将东西放入一个重构的集合中，您甚至可以考虑在数据库之间移动它以解决锁定问题。这仍然可能使您处于需要将其与原始集合交换的位置，但有一些方法可以做到这一点。

---

作为替代方案，您基本上可以直接切入它并立即运行db.eval() 操作。所以这基本上是把流程翻译成对应的JavaScript：

db.eval(function() {

    var lastid = "";
    var counter = 0;
    var text = "";
    var count = [];

    db.collection.find().forEach(function(doc) {
        if ( (doc.passid != lastid) && (counter != 0) ) {
            db.collection.update(
                { "_id": doc._id },
                { 
                    "passid": lastid,
                    "text": text,
                    "count": count
                }
            );
            text = "";
            count = [];
        }
        text = text + " " + doc.text;
        text = text + " " doc.title;
        count.push( doc.count );
        counter++;
        lastid = passid;
    });
})

因此，批量更改文档从来都不是一件好事，但有一些方法可以解决这个问题并将所有操作保留在服务器上。

Answer 2

根据我的经验，使用 mongo 进行此类操作的大部分缓慢来自于数据库的往返，因此请尽量少调用它。如果您的文档足够小（如您的示例所示）以使整个集合适合内存，则可以通过进行一次多次插入和多次删除来节省大量时间：

passids= db.collection.aggregate({ "$group": {"_id": '$passid'}})

new_docs = []

for i in passids['result']:
    doc = {}
    doc['passid'] = i['_id']
    documents = db.collection.find({"passid": i['_id']})
    doc['count'] = []
    doc['text'] = ""

    for d in documents:
        doc['text'] = doc['text'] + " " + d['text']
        doc['text'] = doc['text'] + " " + d['title']
        doc['count'].append(d['count'])

   new_docs.append(doc)

# Instead of removing all the documents one by one, 
# dropping the collection is much faster
db.collection.drop()

db.collection.insert(new_docs)

为了安全起见，我会将文档保存在新集合中，并在检查一切正常后才删除旧集合。