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(); }

Answer 2

我建议您将 $match 设置为管道中的第一个操作，因为 $match 运算符只能在聚合管道中位于第一个时使用索引：

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

如果您还没有定义一个索引，您还应该在 status 和 customer 上添加一个索引：

db.delivery.createIndex({status:1,customer:1})