MongoDB 聚合管道（Aggregation Pipeline） -数据库-火龙果软件工程

捐助

MongoDB 聚合管道（Aggregation Pipeline）

作者：张善友来源：博客园发布于：2015-03-20

次浏览

管道概念

POSIX多线程的使用方式中，有一种很重要的方式-----流水线（亦称为“管道”）方式，“数据元素”流串行地被一组线程按顺序执行。它的使用架构可参考下图：

以面向对象的思想去理解，整个流水线，可以理解为一个数据传输的管道；该管道中的每一个工作线程，可以理解为一个整个流水线的一个工作阶段stage,这些工作线程之间的合作是一环扣一环的。靠输入口越近的工作线程，是时序较早的工作阶段stage,它的工作成果会影响下一个工作线程阶段（stage）的工作结果,即下个阶段依赖于上一个阶段的输出，上一个阶段的输出成为本阶段的输入。这也是pipeline的一个共有特点！

为了回应用户对简单数据访问的需求,MongoDB2.2版本引入新的功能聚合框架（Aggregation Framework），它是数据聚合的一个新框架，其概念类似于数据处理的管道。每个文档通过一个由多个节点组成的管道，每个节点有自己特殊的功能（分组、过滤等），文档经过管道处理后，最后输出相应的结果。管道基本的功能有两个：

一是对文档进行“过滤”，也就是筛选出符合条件的文档;

二是对文档进行“变换”，也就是改变文档的输出形式。

其他的一些功能还包括按照某个指定的字段分组和排序等。而且在每个阶段还可以使用表达式操作符计算平均值和拼接字符串等相关操作。管道提供了一个MapReduce 的替代方案，MapReduce使用相对来说比较复杂，而管道的拥有固定的接口(操作符表达),使用比较简单，对于大多数的聚合任务管道一般来说是首选方法。

该框架使用声明性管道符号来支持类似于SQL Group By操作的功能，而不再需要用户编写自定义的JavaScript例程。

大部分管道操作会在“aggregate”子句后会跟上“$match”打头。它们用在一起，就类似于SQL的from和where子句，或是MongoDB的find函数。“$project”子句看起来也非常类似SQL或MongoDB中的某个概念（和SQL不同的是，它位于表达式尾端）。

接下来介绍的操作在MongoDB聚合框架中是独一无二的。与大多数关系数据库不同，MongoDB天生就可以在行/文档内存储数组。尽管该特性对于全有全无的数据访问十分便利，但是它对于需要组合投影、分组和过滤操作来编写报告的工作，却显得相当复杂。“$unwind”子句将数组分解为单个的元素，并与文档的其余部分一同返回。

“$group”操作与SQL的Group By子句用途相同，但是使用起来却更像是LINQ中的分组运算符。与取回一行平面数据不同，“$group”操作的结果集会呈现为一个持续的嵌套结构。正因如此，使用“$group”可以返回聚合信息，例如对于每个分组中的实际文档，计算文档整体或部分的数目和平均值。

管道操作符

管道是由一个个功能节点组成的，这些节点用管道操作符来进行表示。聚合管道以一个集合中的所有文档作为开始，然后这些文档从一个操作节点流向下一个节点，每个操作节点对文档做相应的操作。这些操作可能会创建新的文档或者过滤掉一些不符合条件的文档，在管道中可以对文档进行重复操作。

先看一个管道聚合的例子：

管道操作符的种类：

管道操作符详细使用说明

1. $project: 数据投影，主要用于重命名、增加和删除字段

例如：

db.article.aggregate(

{ $project : {

title : 1 ,

author : 1 ,

}}

);

这样的话结果中就只还有_id,tilte和author三个字段了，默认情况下_id字段是被包含的，如果要想不包含_id话可以这样:

db.article.aggregate(

{ $project : {

_id : 0 ,

title : 1 ,

author : 1

}});

也可以在$project内使用算术类型表达式操作符，例如：

db.article.aggregate(

{ $project : {

title : 1,

doctoredPageViews : { $add:["$pageViews", 10] }

}});

通过使用$add给pageViews字段的值加10，然后将结果赋值给一个新的字段:doctoredPageViews

注:必须将$add计算表达式放到中括号里面

除此之外使用$project还可以重命名字段名和子文档的字段名:

db.article.aggregate(

{ $project : {

title : 1 ,

page_views : "$pageViews" ,

bar : "$other.foo"

}});

也可以添加子文档：

db.article.aggregate(

{ $project : {

title : 1 ,

stats : {

pv : "$pageViews",

foo : "$other.foo",

dpv : { $add:["$pageViews", 10] }

}

}});

产生了一个子文档stats,里面包含pv,foo,dpv三个字段。

2.$match: 滤波操作，筛选符合条件文档，作为下一阶段的输入

$match的语法和查询表达式(db.collection.find())的语法相同

db.articles.aggregate( [

{ $match : { score : { $gt : 70, $lte : 90 } } },

{ $group: { _id: null, count: { $sum: 1 } } }

] );

$match用于获取分数大于70小于或等于90记录，然后将符合条件的记录送到下一阶段$group管道操作符进行处理。

注意：1.不能在$match操作符中使用$where表达式操作符。

2.$match尽量出现在管道的前面，这样可以提早过滤文档，加快聚合速度。

3.如果$match出现在最前面的话，可以使用索引来加快查询。

3. $limit: 限制经过管道的文档数量

$limit的参数只能是一个正整数

db.article.aggregate(

{ $limit : 5 });

这样的话经过$limit管道操作符处理后，管道内就只剩下前5个文档了

4. $skip: 从待操作集合开始的位置跳过文档的数目

$skip参数也只能为一个正整数

db.article.aggregate(

{ $skip : 5 });

经过$skip管道操作符处理后，前五个文档被“过滤”掉

5.$unwind：将数组元素拆分为独立字段

例如:article文档中有一个名字为tags数组字段：

> db.article.find() 
  { "_id" : ObjectId("528751b0e7f3eea3d1412ce2"),

"author" : "Jone", "title" : "Abook", 

"tags" : [  "good",  "fun",  "good" ] }

使用$unwind操作符后：

> db.article.aggregate({$project:{author:1,title:1,tags:1}},{$unwind:"$tags"}) 
{ 
        "result" : [ 
                { 
                        "_id" : ObjectId("528751b0e7f3eea3d1412ce2"), 
                        "author" : "Jone", 
                        "title" : "A book", 
"tags" : "good" 
                }, 
                { 
                        "_id" : ObjectId("528751b0e7f3eea3d1412ce2"), 
                        "author" : "Jone", 
                        "title" : "A book", 
"tags" : "fun" 
                }, 
                { 
                        "_id" : ObjectId("528751b0e7f3eea3d1412ce2"), 
                        "author" : "Jone", 
                        "title" : "A book", 
  "tags" : "good" 
                } 
        ], 
        "ok" : 1 
}

注意：a.{$unwind:"$tags"})不要忘了$符号

b.如果$unwind目标字段不存在的话，那么该文档将被忽略过滤掉，例如：

 > db.article.aggregate({$project:{author:1,title:1,tags:1}},{$unwind:"$tag"}) 
    { "result" : [ ], "ok" : 1 }

将$tags改为$tag因不存在该字段，该文档被忽略，输出的结果为空

c.如果$unwind目标字段不是一个数组的话，将会产生错误，例如：

> db.article.aggregate({$project:{author:1,title:1,tags:1}},{$unwind:"$title"})

    Error: Printing Stack Trace 
    at printStackTrace (src/mongo/shell/utils.js:37:15) 
    at DBCollection.aggregate (src/mongo/shell/collection.js:897:9) 
    at (shell):1:12 
    Sat Nov 16 19:16:54.488 JavaScript execution failed: aggregate failed: { 
        "errmsg" : "exception: $unwind:  value at end of field path must be an array", 
        "code" : 15978, 
        "ok" : 0 
} at src/mongo/shell/collection.js:L898

d.如果$unwind目标字段数组为空的话，该文档也将会被忽略。

6.$group 对数据进行分组

$group的时候必须要指定一个_id域，同时也可以包含一些算术类型的表达式操作符：

db.article.aggregate(

{ $group : {

_id : "$author",

docsPerAuthor : { $sum : 1 },

viewsPerAuthor : { $sum : "$pageViews" }

}});

注意： 1.$group的输出是无序的。

2.$group操作目前是在内存中进行的，所以不能用它来对大量个数的文档进行分组。

7.$sort : 对文档按照指定字段排序

使用方式如下：

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

按照年龄进行降序操作，按照posts进行升序操作

注意：1.如果将$sort放到管道前面的话可以利用索引，提高效率

2.MongoDB 24.对内存做了优化，在管道中如果$sort出现在$limit之前的话，$sort只会对前$limit个文档进行操作，这样在内存中也只会保留前$limit个文档，从而可以极大的节省内存

3.$sort操作是在内存中进行的，如果其占有的内存超过物理内存的10%，程序会产生错误

8.$goNear

$goNear会返回一些坐标值，这些值以按照距离指定点距离由近到远进行排序

具体使用参数见下表:

例如：

db.places.aggregate([

{

$geoNear: {

near: [40.724, -73.997],

distanceField: "dist.calculated",

maxDistance: 0.008,

query: { type: "public" },

includeLocs: "dist.location",

uniqueDocs: true,

num: 5

}

}

])

其结果为：

{

"result" : [

{ "_id" : 7,

"name" : "Washington Square",

"type" : "public",

"location" : [

[ 40.731, -73.999 ],

[ 40.732, -73.998 ],

[ 40.730, -73.995 ],

[ 40.729, -73.996 ]

],

"dist" : {

"calculated" : 0.0050990195135962296,

"location" : [ 40.729, -73.996 ]

}

},

{ "_id" : 8,

"name" : "Sara D. Roosevelt Park",

"type" : "public",

"location" : [

[ 40.723, -73.991 ],

[ 40.723, -73.990 ],

[ 40.715, -73.994 ],

[ 40.715, -73.994 ]

],

"dist" : {

"calculated" : 0.006082762530298062,

"location" : [ 40.723, -73.991 ]

}

}

],

"ok" : 1}

其中，dist.calculated中包含了计算的结果，而dist.location中包含了计算距离时实际用到的坐标

注意： 1.使用$goNear只能在管道处理的开始第一个阶段进行

2.必须指定distanceField，该字段用来决定是否包含距离字段

3.$gonNear和geoNear命令比较相似，但是也有一些不同:distanceField在$geoNear中是必选的，而在geoNear中是可选的；includeLocs在$geoNear中是string类型，而在geoNear中是boolen类型。

管道表达式

管道操作符作为“键”,所对应的“值”叫做管道表达式。例如上面例子中{$match:{status:"A"}}，$match称为管道操作符，而{status:"A"}称为管道表达式，它可以看作是管道操作符的操作数(Operand)，每个管道表达式是一个文档结构，它是由字段名、字段值、和一些表达式操作符组成的，例如上面例子中管道表达式就包含了一个表达式操作符$sum进行累加求和。

每个管道表达式只能作用于处理当前正在处理的文档，而不能进行跨文档的操作。管道表达式对文档的处理都是在内存中进行的。除了能够进行累加计算的管道表达式外，其他的表达式都是无状态的，也就是不会保留上下文的信息。累加性质的表达式操作符通常和$group操作符一起使用，来统计该组内最大值、最小值等，例如上面的例子中我们在$group管道操作符中使用了具有累加的$sum来计算总和。

除了$sum以为，还有以下性质的表达式操作符：

组聚合操作符