mongodb - 聚合和更新 MongoDB

Question

我有 2 个集合:

• 客户(6 000 000 份文档)
• 订单(50 000 000 份文档)

每天一次，我想按客户计算过去一年、过去一个月和过去一周等的订单数。

我试过这个:

db.orders.aggregate(
    {$match: 
        { date_order: { $gt: v_date1year } }
    },
    {$group : {
        _id : "$id_client", 
        count : {$sum : 1}
    }} ,
    {
        "$out": "tmp_indicators"
    }
)

db.tmp_indicators.find({}).forEach(function (my_client) { 
    db.clients.update (
        {"id_client": my_client._id},
        {"$set": 
            { "nb_orders_1year" : my_client.count }
        }
    )
})

我必须这样做 3 次，1 次用于过去一年的聚合，1 次用于过去一个月，1 次用于过去一周。处理速度很慢，您知道如何以更好的方式执行吗？

Answer 1

为了提高性能，尤其是在处理大型集合时，请利用 Bulk() 用于批量更新的 API，因为您将分批将操作发送到服务器(例如，批量大小为 1000)，这会给您带来更好的性能，因为您不会将每个请求都发送到服务器(正如您当前在 forEach() 循环中使用更新语句所做的那样)，但每 1000 个请求只执行一次，从而使您的更新比当前更高效、更快速。

以下示例演示了这种方法，第一个使用 Bulk() API 在 MongoDB 版本中可用 >= 2.6 and < 3.2 .它更新了 clients 中的所有文档通过更改 nb_orders_1year 进行收集具有聚合结果值的字段。

自 aggregate() 方法返回 cursor , 可以使用聚合输出集合的 forEach() 迭代它并访问每个文档的方法，从而分批设置批量更新操作，然后使用 API 有效地跨服务器发送:

var bulk = db.clients.initializeUnorderedBulkOp(),
    pipeline = [
        {
            "$match": { "date_order": { "$gt": v_date1year } }
        },
        {
            "$group": {
                "_id": "$id_client", 
                "count": { "$sum" : 1 }
            }
        },
        { "$out": "tmp_indicators" }        
    ],
    counter = 0;

db.orders.aggregate(pipeline);  
db.tmp_indicators.find().forEach(function (doc) {       
    bulk.find({ "_id": doc._id }).updateOne({ 
        "$set": { "nb_orders_1year": doc.count }
    });

    counter++;
    if (counter % 1000 == 0) {
        bulk.execute(); // Execute per 1000 operations and re-initialize every 1000 update statements
        bulk = db.clients.initializeUnorderedBulkOp();
    }
});
// Clean up remaining operations in queue
if (counter % 1000 != 0) { bulk.execute(); }

下一个例子适用于新的 MongoDB 版本 3.2自deprecated the Bulk API以来并使用 bulkWrite() 提供了一组较新的 api .

它使用与上面相同的游标，但不是迭代结果，而是使用其 map() 创建具有批量操作的数组方法:

 var pipeline = [
        {
            "$match": { "date_order": { "$gt": v_date1year } }
        },
        {
            "$group": {
                "_id": "$id_client", 
                "count": { "$sum" : 1 }
            }
        },
        { "$out": "tmp_indicators" }        
    ];
db.orders.aggregate(pipeline);
var bulkOps = db.tmp_indicators.find().map(function (doc) { 
        return { 
            "updateOne": { 
                "filter": { "_id": doc._id } ,              
                "update": { "$set": { "nb_orders_1year": doc.count } } 
            }         
        };
    });

db.clients.bulkWrite(bulkOps, { "ordered": true });