组内MongoDB聚合排序和限制[重复]

Question

我有一个具有以下模式的待售项目集合：

var itemSchema = new Schema({
    "category" : { type : Schema.Types.ObjectId, ref : 'Category' },
    "merchant" : { type : Schema.Types.ObjectId, ref : 'Merchant' },
    "rating" : Number
})

我继承了一个聚合查询，它返回匹配类别的项目，按商家分组，组按组中的最大评级排序：

Item.aggregate([
      { "$match" : { category : categoryId, merchant : { $ne : null }}},
      { "$group" : { _id : "$merchant", 
                    rating : { $max : "$rating" }, 
                    items : { $push : "$$ROOT" }}},
      { "$sort" : { rating : -1 }}
    ], { allowDiskUse : true })
    .skip(skip)
    .limit(limit)

在此之后，代码继续按评级对每组中的项目进行排序，并删除除每组中排名前2位之外的所有项目。

作为聚合函数的一部分，是否可以在组内执行此排序和限制，以便聚合只返回每组中评级最高的两个项目？

Answer 1

我所做的是只推送id，然后添加切片。

$group: {
    _id: "$merchant",
    items : { $push : "$_id" }
},
// ...

$project: {
  items: {$slice: ["$items", offset, limit]}
}

在查询之后，我填充结果：

return Item.populate(result, {path: 'items'})

Answer 2

在可预见的不久的将来，在当前的聚合框架中尝试这样做并不是最明智的想法。主要问题当然来自您已经拥有的代码中的这一行：

"items" : { "$push": "$$ROOT" }

这就意味着，基本上需要发生的是，分组键中的所有对象都需要被放入一个数组中，以便在以后的代码中得到“top N”结果。

这显然无法扩展，因为该阵列本身的大小可能会超过16MB的BSON限制，并且无法覆盖分组文档中的其余数据。这里的主要问题是，不可能“限制推送”仅限于一定数量的物品。在这件事上，吉拉的问题由来已久。

仅出于这个原因，最实用的方法就是为每个分组键的“前N个”项运行单独的查询。这些甚至不需要是。aggregate（）stations（取决于数据），实际上可以是任何只限制您想要的“前N个”值的东西。

您的体系结构似乎位于节点上。js和mongoose，但是任何支持异步IO和并行执行查询的东西都是最好的选择。理想情况下，它有自己的API库，支持将这些查询的结果组合到单个响应中。

例如，有一个使用您的架构和可用库（特别是async）的简化示例列表，它完全执行并行和组合结果：

var async = require('async'),
    mongoose = require('mongoose'),
    Schema = mongoose.Schema;

mongoose.connect('mongodb://localhost/test');

var data = [
  { "merchant": 1, "rating": 1 },
  { "merchant": 1, "rating": 2 },
  { "merchant": 1, "rating": 3 },
  { "merchant": 2, "rating": 1 },
  { "merchant": 2, "rating": 2 },
  { "merchant": 2, "rating": 3 }
];

var testSchema = new Schema({
  merchant: Number,
  rating: Number
});

var Test = mongoose.model( 'Test', testSchema, 'test' );

async.series(
  [
    function(callback) {
      Test.remove({},callback);
    },
    function(callback) {
      async.each(data,function(item,callback) {
        Test.create(item,callback);
      },callback);
    },
    function(callback) {
      async.waterfall(
        [
          function(callback) {
            Test.distinct("merchant",callback);
          },
          function(merchants,callback) {
            async.concat(
              merchants,
              function(merchant,callback) {
                Test.find({ "merchant": merchant })
                  .sort({ "rating": -1 })
                  .limit(2)
                  .exec(callback);
              },
              function(err,results) {
                console.log(JSON.stringify(results,undefined,2));
                callback(err);
              }
            );
          }
        ],
        callback
      );
    }
  ],
  function(err) {
    if (err) throw err;
    mongoose.disconnect();
  }
);

这会导致输出中每个商家的前2个结果：

[
  {
    "_id": "560d153669fab495071553ce",
    "merchant": 1,
    "rating": 3,
    "__v": 0
  },
  {
    "_id": "560d153669fab495071553cd",
    "merchant": 1,
    "rating": 2,
    "__v": 0
  },
  {
    "_id": "560d153669fab495071553d1",
    "merchant": 2,
    "rating": 3,
    "__v": 0
  },
  {
    "_id": "560d153669fab495071553d0",
    "merchant": 2,
    "rating": 2,
    "__v": 0
  }
]

这确实是处理此问题的最有效方法，尽管它会占用资源，因为它仍然是多个查询。但是如果您尝试将所有文档存储在数组中并对其进行处理，则与聚合管道中消耗的资源相去甚远。

因此，考虑到文件数量不会导致BSON限制的违反，可以这样做。MongoDB当前版本中的方法对此并不适用，但即将发布的版本（截至编写之时，3.1.8 dev branch会这样做）至少在聚合管道中引入了一个$slice操作符。因此，如果您在聚合操作方面更聪明，并且首先使用排序$，那么就可以轻松地挑出数组中已排序的项目：

var async = require('async'),
    mongoose = require('mongoose'),
    Schema = mongoose.Schema;

mongoose.connect('mongodb://localhost/test');

var data = [
  { "merchant": 1, "rating": 1 },
  { "merchant": 1, "rating": 2 },
  { "merchant": 1, "rating": 3 },
  { "merchant": 2, "rating": 1 },
  { "merchant": 2, "rating": 2 },
  { "merchant": 2, "rating": 3 }
];

var testSchema = new Schema({
  merchant: Number,
  rating: Number
});

var Test = mongoose.model( 'Test', testSchema, 'test' );

async.series(
  [
    function(callback) {
      Test.remove({},callback);
    },
    function(callback) {
      async.each(data,function(item,callback) {
        Test.create(item,callback);
      },callback);
    },
    function(callback) {
      Test.aggregate(
        [
          { "$sort": { "merchant": 1, "rating": -1 } },
          { "$group": {
            "_id": "$merchant",
            "items": { "$push": "$$ROOT" }
          }},
          { "$project": {
            "items": { "$slice": [ "$items", 2 ] }
          }}
        ],
        function(err,results) {
          console.log(JSON.stringify(results,undefined,2));
          callback(err);
        }
      );
    }
  ],
  function(err) {
    if (err) throw err;
    mongoose.disconnect();
  }
);

这产生的基本结果与前2个项目在首先排序后从数组中“切片”相同。

在当前版本中，这实际上也是“可能的”，但有相同的基本限制，即这仍然需要在对内容进行排序后将所有内容推送到一个数组中。它只需要一种“迭代”的方法。您可以将其编码出来，为更多的条目生成聚合管道，但仅显示“两个”应该表明尝试以下操作并不是一个好主意：

var async = require('async'),
    mongoose = require('mongoose'),
    Schema = mongoose.Schema;

mongoose.connect('mongodb://localhost/test');

var data = [
  { "merchant": 1, "rating": 1 },
  { "merchant": 1, "rating": 2 },
  { "merchant": 1, "rating": 3 },
  { "merchant": 2, "rating": 1 },
  { "merchant": 2, "rating": 2 },
  { "merchant": 2, "rating": 3 }
];

var testSchema = new Schema({
  merchant: Number,
  rating: Number
});

var Test = mongoose.model( 'Test', testSchema, 'test' );

async.series(
  [
    function(callback) {
      Test.remove({},callback);
    },
    function(callback) {
      async.each(data,function(item,callback) {
        Test.create(item,callback);
      },callback);
    },
    function(callback) {
      Test.aggregate(
        [
          { "$sort": { "merchant": 1, "rating": -1 } },
          { "$group": {
            "_id": "$merchant",
            "items": { "$push": "$$ROOT" }
          }},
          { "$unwind": "$items" },
          { "$group": {
            "_id": "$_id",
            "first": { "$first": "$items" },
            "items": { "$push": "$items" }
          }},
          { "$unwind": "$items" },
          { "$redact": {
            "$cond": [
              { "$eq": [ "$items", "$first" ] },
              "$$PRUNE",
              "$$KEEP"
            ]
          }},
          { "$group": {
            "_id": "$_id",
            "first": { "$first": "$first" },
            "second": { "$first": "$items" }
          }},
          { "$project": {
            "items": {
              "$map": {
                "input": ["A","B"],
                "as": "el",
                "in": {
                  "$cond": [
                    { "$eq": [ "$$el", "A" ] },
                    "$first",
                    "$second"
                  ]
                }
              }
            }
          }}
        ],
        function(err,results) {
          console.log(JSON.stringify(results,undefined,2));
          callback(err);
        }
      );
    }
  ],
  function(err) {
    if (err) throw err;
    mongoose.disconnect();
  }
);

尽管在早期版本中“可能”（这是使用2.6引入的功能来缩短，因为您已经标记了根目录），基本步骤是存储数组，然后使用$first将每个项目“从堆栈中取出”，并将其（以及可能的其他项目）与数组中的项目进行比较以删除它们，然后从堆栈中取出“下一个第一个”项目，直到最终完成“前N个”。

直到有一天有这样一个操作允许$ush聚合累加器中的项目被限制在某个计数，那么这对于聚合来说并不是一个真正的实际操作。

如果您在这些结果中拥有的数据足够小，您可以做到这一点，如果数据库服务器具有足够的规格以提供真正的优势，它甚至可能比客户端处理更有效。但在大多数合理使用的实际应用程序中，这两种情况都有可能。

最好的选择是使用首先演示的“并行查询”选项。它总是能够很好地扩展，并且不需要“编写”这样的逻辑，即特定分组可能不会返回至少所需的总“前N”项，并且在它只执行每个查询并组合结果时，确定如何保留它们（省略了更长的示例）。

使用并行查询。它将比您现有的编码方法更好，而且它将在很大程度上优于聚合方法。至少在有更好的选择之前。