mongodb聚合管道用法

基本用法

db.collection.aggregate( [ { <stage> }, ... ] )

stage如下

名称	描述
$addFields	将新的字段添加到文档中，输出的文档包含已经存在的字段和新加入的字段
$bucket	根据指定的表达式和存储区边界将传入文档分组到称为buckets的组中。
$bucketAuto	根据指定的表达式将传入文档分类到特定数量的组（称为buckets）。存储区边界自动确定，试图将文档均匀分布到指定数量的buckets中。
$collStats	返回有关集合或视图的统计信息。
$count	返回聚合管道的计数
$currentOp	返回有关MongoDB部署的活动和/或休眠操作的信息
$facet	在同一组输入文档中的单个阶段内处理多个聚合流水线。支持创建多方面的聚合，能够在单个阶段中跨多个维度或方面表征数据。
$geoNear	根据地理空间点的接近度返回有序的文档流。包含地理空间数据的$ match，$ sort和$ limit功能。输出文件包含一个额外的距离字段，并可包含位置标识符字段。
$graphLookup	对集合执行递归搜索。为每个输出文档添加一个新的数组字段，其中包含该文档的递归搜索的遍历结果
$group	按指定的标识符表达式输入文档，并将累加器表达式（如果指定）应用于每个组。消耗所有输入文档并为每个不同的组输出一个文档。输出文件只包含标识符字段，如果指定了，则包含累积字段。
$indexStats	返回有关使用集合中每个索引的统计信息。
$limit	将未修改的前n个文档传递到管道，其中n是指定的限制。对于每个输入文档，输出一个文档（前n个文档）或零文档（前n个文档之后）。
$listLocalSessions	列出最近在当前连接的mongos或mongod实例中使用的所有活动会话。这些会话可能尚未传播到system.sessions集合。
$listSessions	列出所有活动时间足以传播到system.sessions集合的所有会话。
$lookup	将左外连接执行到同一数据库中的另一个集合，以过滤“已连接”集合中的文档进行处理。
$match	过滤文档流，只允许匹配的文档未经修改地传递到下一个管道阶段。 $ match使用标准的MongoDB查询。对于每个输入文档，输出一个文档（匹配）或零个文档（不匹配）。
$out	将聚合管道的结果文档写入集合。要使用$ out阶段，它必须是管道中的最后一个阶段。
$project	重新设计流中的每个文档，例如添加新字段或删除现有字段。对于每个输入文档，输出一个文档。
$redact	根据存储在文档本身中的信息限制每个文档的内容，重新整形流中的每个文档。包含$ project和$ match的功能。可用于实施字段级别的编校。对于每个输入文档，输出一个或零个文档。
$replaceRoot	用指定的嵌入式文档替换文档。该操作将替换输入文档中的所有现有字段，包括_id字段。指定嵌入在输入文档中的文档以将嵌入式文档提升到顶层。
$sample	从其输入中随机选择指定数量的文档。
$skip	跳过前n个文档，其中n是指定的跳过编号，并将未修改的其余文档传递到管道。对于每个输入文档，输出零文档（对于前n个文档）或一个文档（如果在前n个文档之后）。
$sort	通过指定的排序键对文档流进行重新排序。只有订单改变了;文件保持不变。对于每个输入文档，输出一个文档。
$sortByCount	根据指定表达式的值对传入文档分组，然后计算每个不同组中文档的数量。
$unwind	从输入文档解构数组字段以输出每个元素的文档。每个输出文档用一个元素值替换数组。对于每个输入文档，输出n个文档，其中n是数组元素的数量，对于空数组可以为零。

以上翻译自谷歌翻译

常用的stage有

$count

经常搭配其他的stage一起使用

样例

假设数据如下

{ "_id" : 1, "subject" : "History", "score" : 88 }
{ "_id" : 2, "subject" : "History", "score" : 92 }
{ "_id" : 3, "subject" : "History", "score" : 97 }
{ "_id" : 4, "subject" : "History", "score" : 71 }
{ "_id" : 5, "subject" : "History", "score" : 79 }
{ "_id" : 6, "subject" : "History", "score" : 83 }

使用以下聚合操作

db.scores.aggregate(
  [
    {
      $match: {
        score: {
          $gt: 80
        }
      }
    },
    {
      $count: "passing_scores"
    }
  ]
)

意思是匹配score字段大于80分的文档，然后计算数量，重命名为passing_scores输出

输出如下

{ "passing_scores" : 4 }

$group

用法如下

{ $group: { _id: <expression>, <field1>: { <accumulator1> : <expression1> }, ... } }

<accumulator>操作如下

$sum,$avg,$first,$last,$max,$min,$push,$addToSet,$stdDevPop,$stdDevSamp

用途就跟名称差不多

样例

{ "_id" : 1, "item" : "abc", "price" : 10, "quantity" : 2, "date" : ISODate("2014-03-01T08:00:00Z") }
{ "_id" : 2, "item" : "jkl", "price" : 20, "quantity" : 1, "date" : ISODate("2014-03-01T09:00:00Z") }
{ "_id" : 3, "item" : "xyz", "price" : 5, "quantity" : 10, "date" : ISODate("2014-03-15T09:00:00Z") }
{ "_id" : 4, "item" : "xyz", "price" : 5, "quantity" : 20, "date" : ISODate("2014-04-04T11:21:39.736Z") }
{ "_id" : 5, "item" : "abc", "price" : 10, "quantity" : 10, "date" : ISODate("2014-04-04T21:23:13.331Z") }

使用以下聚合操作

db.sales.aggregate(
   [
      {
        $group : {
           _id : { month: { $month: "$date" }, day: { $dayOfMonth: "$date" }, year: { $year: "$date" } },
           totalPrice: { $sum: { $multiply: [ "$price", "$quantity" ] } },
           averageQuantity: { $avg: "$quantity" },
           count: { $sum: 1 }
        }
      }
   ]
)

意思是将文档根据时间(相同的年月日)分组，并计算其总价格(价格*数量求和)，平均的数量，每个分组有多少个文档

结果如下

{ "_id" : { "month" : 3, "day" : 15, "year" : 2014 }, "totalPrice" : 50, "averageQuantity" : 10, "count" : 1 }
{ "_id" : { "month" : 4, "day" : 4, "year" : 2014 }, "totalPrice" : 200, "averageQuantity" : 15, "count" : 2 }
{ "_id" : { "month" : 3, "day" : 1, "year" : 2014 }, "totalPrice" : 40, "averageQuantity" : 1.5, "count" : 2 }

使用以下聚合操作

db.sales.aggregate(
   [
      {
        $group : {
           _id : null,
           totalPrice: { $sum: { $multiply: [ "$price", "$quantity" ] } },
           averageQuantity: { $avg: "$quantity" },
           count: { $sum: 1 }
        }
      }
   ]
)

结果如下

{ "_id" : null, "totalPrice" : 290, "averageQuantity" : 8.6, "count" : 5 }

更多操作见https://docs.mongodb.com/manual/reference/operator/aggregation/group/#pipe._S_group

$limit

用法如下

db.article.aggregate(
    { $limit : 5 }
);

意思就是现在article只返回前5个文档

$lookup

用法如下

{
   $lookup:
     {
       from: <collection to join>,
       localField: <field from the input documents>,
       foreignField: <field from the documents of the "from" collection>,
       as: <output array field>
     }
}

from表示另一个要连接的表b，localField表示该表a中用于和表b连接的字段，foreignField表示表b中用于和表a连接的字段，as表示为输出结果命名

SELECT *, <output array field>
FROM collection
WHERE <output array field> IN (SELECT *
                               FROM <collection to join>
                               WHERE <foreignField>= <collection.localField>);

相当于以上的sql语句

另一种用法

{
   $lookup:
     {
       from: <collection to join>,
       let: { <var_1>: <expression>, …, <var_n>: <expression> },
       pipeline: [ <pipeline to execute on the collection to join> ],
       as: <output array field>
     }
}

对应的sql语句

SELECT *, <output array field>
FROM collection
WHERE <output array field> IN (SELECT <documents as determined from the pipeline>
                               FROM <collection to join>
                               WHERE <pipeline> );

样例

插入如下数据

db.orders.insert([
   { "_id" : 1, "item" : "almonds", "price" : 12, "quantity" : 2 },
   { "_id" : 2, "item" : "pecans", "price" : 20, "quantity" : 1 },
   { "_id" : 3  }
])
db.inventory.insert([
   { "_id" : 1, "sku" : "almonds", description: "product 1", "instock" : 120 },
   { "_id" : 2, "sku" : "bread", description: "product 2", "instock" : 80 },
   { "_id" : 3, "sku" : "cashews", description: "product 3", "instock" : 60 },
   { "_id" : 4, "sku" : "pecans", description: "product 4", "instock" : 70 },
   { "_id" : 5, "sku": null, description: "Incomplete" },
   { "_id" : 6 }
])

使用以下聚合操作

db.orders.aggregate([
   {
     $lookup:
       {
         from: "inventory",
         localField: "item",
         foreignField: "sku",
         as: "inventory_docs"
       }
  }
])

返回结果如下

{
   "_id" : 1,
   "item" : "almonds",
   "price" : 12,
   "quantity" : 2,
   "inventory_docs" : [
      { "_id" : 1, "sku" : "almonds", "description" : "product 1", "instock" : 120 }
   ]
}
{
   "_id" : 2,
   "item" : "pecans",
   "price" : 20,
   "quantity" : 1,
   "inventory_docs" : [
      { "_id" : 4, "sku" : "pecans", "description" : "product 4", "instock" : 70 }
   ]
}
{
   "_id" : 3,
   "inventory_docs" : [
      { "_id" : 5, "sku" : null, "description" : "Incomplete" },
      { "_id" : 6 }
   ]
}

相同sql语句如下

SELECT *, inventory_docs
FROM orders
WHERE inventory_docs IN (SELECT *
FROM inventory
WHERE sku= orders.item);

更多内容见https://docs.mongodb.com/manual/reference/operator/aggregation/lookup/#pipe._S_lookup

$match

用法如下

{ $match: { <query> } }

样例

数据如下

{ "_id" : ObjectId("512bc95fe835e68f199c8686"), "author" : "dave", "score" : 80, "views" : 100 }
{ "_id" : ObjectId("512bc962e835e68f199c8687"), "author" : "dave", "score" : 85, "views" : 521 }
{ "_id" : ObjectId("55f5a192d4bede9ac365b257"), "author" : "ahn", "score" : 60, "views" : 1000 }
{ "_id" : ObjectId("55f5a192d4bede9ac365b258"), "author" : "li", "score" : 55, "views" : 5000 }
{ "_id" : ObjectId("55f5a1d3d4bede9ac365b259"), "author" : "annT", "score" : 60, "views" : 50 }
{ "_id" : ObjectId("55f5a1d3d4bede9ac365b25a"), "author" : "li", "score" : 94, "views" : 999 }
{ "_id" : ObjectId("55f5a1d3d4bede9ac365b25b"), "author" : "ty", "score" : 95, "views" : 1000 }

聚合操作如下

db.articles.aggregate(
    [ { $match : { author : "dave" } } ]
);

结果如下

{ "_id" : ObjectId("512bc95fe835e68f199c8686"), "author" : "dave", "score" : 80, "views" : 100 }
{ "_id" : ObjectId("512bc962e835e68f199c8687"), "author" : "dave", "score" : 85, "views" : 521 }

$out

用法如下

{ $out: "<output-collection>" }

样例

数据如下

{ "_id" : 8751, "title" : "The Banquet", "author" : "Dante", "copies" : 2 }
{ "_id" : 8752, "title" : "Divine Comedy", "author" : "Dante", "copies" : 1 }
{ "_id" : 8645, "title" : "Eclogues", "author" : "Dante", "copies" : 2 }
{ "_id" : 7000, "title" : "The Odyssey", "author" : "Homer", "copies" : 10 }
{ "_id" : 7020, "title" : "Iliad", "author" : "Homer", "copies" : 10 }

聚合操作如下

db.books.aggregate( [
                      { $group : { _id : "$author", books: { $push: "$title" } } },
                      { $out : "authors" }
                  ] )

意思是将books的author字段分类并作为_id字段的内容，将相同作者的title都push到books字段中，将其插入authors表中

结果如下

{ "_id" : "Homer", "books" : [ "The Odyssey", "Iliad" ] }
{ "_id" : "Dante", "books" : [ "The Banquet", "Divine Comedy", "Eclogues" ] }

$project

用法如下

{ $project: { <specification(s)> } }

样例

数据如下

{
  "_id" : 1,
  title: "abc123",
  isbn: "0001122223334",
  author: { last: "zzz", first: "aaa" },
  copies: 5
}

聚合操作如下

db.books.aggregate( [ { $project : { title : 1 , author : 1 } } ] )

结果如下

{ "_id" : 1, "title" : "abc123", "author" : { "last" : "zzz", "first" : "aaa" } }

更多内容见https://docs.mongodb.com/manual/reference/operator/aggregation/project/#pipe._S_project

$skip

用法如下

{ $skip: <positive integer> }

样例

db.article.aggregate(
    { $skip : 5 }
);

意思也很简单就是跳过前5个文档

$sort

用法如下

{ $sort: { <field1>: <sort order>, <field2>: <sort order> ... } }

样例

db.users.aggregate(
   [
     { $sort : { age : -1, posts: 1 } }
   ]
)

-1表示降序排列，1表示升序排列

$sortByCont

用法见https://docs.mongodb.com/manual/reference/operator/aggregation/sortByCount/#pipe._S_sortByCount

$unwind

用法见https://docs.mongodb.com/manual/reference/operator/aggregation/unwind/#pipe._S_unwind