node.js - Mongo 聚合查询在服务器上花费的时间太长

Question

我正在尝试在 mongo 中使用聚合框架来获取一些数据统计信息。我正在使用的查询，当在本地运行时几乎不需要一分钟，但是当我在服务器上运行相同的查询时它没有给出响应并且在等待太久之后，我不得不取消它，任何人都可以请建议为什么是这发生了。

var orderIds = db.delivery.find({"status":"DELIVERED"}).map(function(o) {
    return o.order 
}); 

var userIds =  db.order.aggregate([{
    $match : { _id : { $in : orderIds } }
}, {
    $group: { _id : "$customer" }
}]).map(function(u) { return u._id });

var userstats = db.order.aggregate([{
    $sort : { customer : 1, dateCreated : 1 }
}, {
    $match : { status : "DELIVERED", customer : {  $in : userIds } }
}, { 
    $group: {
        _id : "$customer", orders : { $sum : 1 }, 
        firstOrderDate : { $first : "$dateCreated" },
        lastOrderDate : { $last : "$dateCreated" }
    }
}]);

userstats.forEach(function(x) { 
    db.user.update({ _id : x._id }, {
        $set : { 
            totalOrders : x.orders,
            firstOrderDate : x.firstOrderDate,
            lastOrderDate : x.lastOrderDate
        }
    })
})

我不确定，但它在服务器上不应该更快吗？ ，但它无法提供输出。

Answer 1

要加快流程，您可以通过多种方式重构您的操作。 第一个是消除不必要的管道操作，如 $sort 运算符可以替换为 $max 和 $min $group 中的运算符 管道。

其次，使用 bulk() API 这将提高更新操作的性能，尤其是在处理大型集合时，因为它们会将操作分批发送到服务器(例如，批量大小为 500)，这与将每个请求发送到服务器(如您当前正在 forEach() 循环中执行更新语句。

考虑以下重构操作:

var orderIds = db.delivery.find({"status": "DELIVERED"}).map(function(d){return d.order;}),
    counter = 0,
    bulk = db.user.initializeUnorderedBulkOp();

var userstatsCursor = db.orders.aggregate([
    { "$match": { "_id": { "$in": orderIds } } },
    { 
        "$group": { 
            "_id": "$customer", 
            "orders": { "$sum": 1 },
            "firstOrderDate": { "$min": "$dateCreated" },
            "lastOrderDate":{ "$max": "$dateCreated" } } 
        } 
    }
]);

userstatsCursor.forEach(function (x){
    bulk.find({ "_id": x._id }).updateOne({ 
        "$set": { 
            "totalOrders": x.orders,
            "firstOrderDate": x.firstOrderDate,
            "lastOrderDate": x.lastOrderDate
        }
    });

    counter++;
    if (counter % 500 == 0) {
        bulk.execute(); // Execute per 500 operations and 
        // re-initialize every 500 update statements
        bulk = db.user.initializeUnorderedBulkOp();
    }
});

// Clean up remaining operations in queue
if (counter % 500 != 0) { bulk.execute(); }