MongoDB 聚合管道(一)（Aggregation Pipeline）

一 管道的概念
        管道是MongoDB2.2版本引入新的功能 ，它是数据聚合的一个新框架，其概念类似于数据处理的管道。管道的工作方式类似于UNIX-like的shell ps aux | grep mong*  （下面的补充会详细介绍管道原理）。
       每个文档通过一个由多个节点组成的管道，每个节点有自己特殊的功能（分组、过滤等），文档经过管道处理后，最后输出相应的结果。管道基本的功能有两个：一是对文档进行“过滤”，也就是筛选出符合条件的文档;二是对文档进行“变换”，也就是改变文档的输出形式。其他的一些功能还包括按照某个指定的字段分组和排序等。而且在每个阶段还可以使用表达式操作符计算平均值和拼接字符串等相关操作。管道提供了一个MapReduce 的替代方案，MapReduce使用相对来说比较复杂，而管道拥有固定的接口(操作符表达),使用比较简单，对于大多数的聚合任务管道一般来说是首选方法。       二 管道操作符
        管道是由一个个功能节点组成的，这些节点用管道操作符来进行表示。聚合管道以一个集合中的所有文档作为开始，然后这些文档从一个操作节点流向下一个节点 ，每个操作节点对文档做相应的操作。这些操作可能会创建新的文档或者过滤掉一些不符合条件的文档，在管道中可以对文档进行重复操作。         先看一个管道聚合的例子：
                     

       管道操作符的种类：

Name	Description
$project	Reshapes a document stream. $project can rename, add, or remove fields as well as create computed values and sub-documents.
$match	Filters the document stream, and only allows matching documents to pass into the next pipeline stage.$match uses standard MongoDB queries.
$limit	Restricts the number of documents in an aggregation pipeline.
$skip	Skips over a specified number of documents from the pipeline and returns the rest.
$unwind	Takes an array of documents and returns them as a stream of documents.
$group	Groups documents together for the purpose of calculating aggregate values based on a collection of documents.
$sort	Takes all input documents and returns them in a stream of sorted documents.
$geoNear	Returns an ordered stream of documents based on proximity to a geospatial point.

管道操作符详细使用说明
  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会返回一些坐标值，这些值以按照距离指定点距离由近到远进行排序

具体使用参数见下表:

Field	Type	Description
near	GeoJSON point orlegacy coordinate pairs	The point for which to find the closest documents.
distanceField	string	The output field that contains the calculated distance. To specify a field within a subdocument, use dot notation.
limit	number	Optional. The maximum number of documents to return. The default value is 100. See also the num option.
num	number	Optional. The num option provides the same function as the limitoption. Both define the maximum number of documents to return. If both options are included, the num value overrides the limit value.
maxDistance	number	Optional. A distance from the center point. Specify the distance in radians. MongoDB limits the results to those documents that fall within the specified distance from the center point.
query	document	Optional. Limits the results to the documents that match the query. The query syntax is the usual MongoDB read operation query syntax.
spherical	Boolean	Optional. If true, MongoDB references points using a spherical surface. The default value is false.
distanceMultiplier	number	Optional. The factor to multiply all distances returned by the query. For example, use the distanceMultiplier to convert radians, as returned by a spherical query, to kilometers by multiplying by the radius of the Earth.
includeLocs	string	Optional. This specifies the output field that identifies the location used to calculate the distance. This option is useful when a location field contains multiple locations. To specify a field within a subdocument, usedot notation.
uniqueDocs	Boolean	Optional. If this value is true, the query returns a matching document once, even if more than one of the document’s location fields match the query. If this value is false, the query returns a document multiple times if the document has multiple matching location fields. See $uniqueDocsfor more information.

   例如：

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称为管道操作符，而{s tatus:"A"}称为管道表达式，它可以看做是管道操作符的操作数(Operand)，每个管道表达式是一个文档结构，它是由字段名、字段值、和一些表达式操作符组成的，例如上面例子中管道表达式就包含了一个表达式操作符$sum进行累加求和。

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

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

Name	Description
$addToSet	Returns an array of all the unique values for the selected field among for each document in that group.
$first	Returns the first value in a group.
$last	Returns the last value in a group.
$max	Returns the highest value in a group.
$min	Returns the lowest value in a group.
$avg	Returns an average of all the values in a group.
$push	Returns an array of all values for the selected field among for each document in that group.
$sum	Returns the sum of all the values in a group.

Bool类型聚合操作符

Name	Description
$and	Returns true only when all values in its input array are true.
$or	Returns true when any value in its input array are true.
$not	Returns the boolean value that is the opposite of the input value.

比较类型聚合操作符

Name	Description
$cmp	Compares two values and returns the result of the comparison as an integer.
$eq	Takes two values and returns true if the values are equivalent.
$gt	Takes two values and returns true if the first is larger than the second.
$gte	Takes two values and returns true if the first is larger than or equal to the second.
$lt	Takes two values and returns true if the second value is larger than the first.
$lte	Takes two values and returns true if the second value is larger than or equal to the first.
$ne	Takes two values and returns true if the values are not equivalent.

算术类型聚合操作符

Name	Description
$add	Computes the sum of an array of numbers.
$divide	Takes two numbers and divides the first number by the second.
$mod	Takes two numbers and calcualtes the modulo of the first number divided by the second.
$multiply	Computes the product of an array of numbers.
$subtract	Takes two numbers and subtracts the second number from the first.

字符串类型聚合操作符

Name	Description
$concat	Concatenates two strings.
$strcasecmp	Compares two strings and returns an integer that reflects the comparison.
$substr	Takes a string and returns portion of that string.
$toLower	Converts a string to lowercase.
$toUpper	Converts a string to uppercase.

日期类型聚合操作符

Name	Description
$dayOfYear	Converts a date to a number between 1 and 366.
$dayOfMonth	Converts a date to a number between 1 and 31.
$dayOfWeek	Converts a date to a number between 1 and 7.
$year	Converts a date to the full year.
$month	Converts a date into a number between 1 and 12.
$week	Converts a date into a number between 0 and 53
$hour	Converts a date into a number between 0 and 23.
$minute	Converts a date into a number between 0 and 59.
$second	Converts a date into a number between 0 and 59. May be 60 to account for leap seconds.
$millisecond	Returns the millisecond portion of a date as an integer between 0 and 999.

条件类型聚合操作符

Name	Description
$cond	A ternary operator that evaluates one expression, and depending on the result returns the value of one following expressions.
$ifNull	Evaluates an expression and returns a value.

注：以上操作符都必须在管道操作符的表达式内来使用。
各个表达式操作符的具体使用方式参见： http://docs.mongodb.org/manual/reference/operator/aggregation-group/

四 总结        管道是MongoDB新引入的聚合框架，它使用了大量的管道操作符来固定聚合的接口，每种管道操作符中又可以包含丰富的表达式操作符，每种操作符都有其特定的功能，只有在理解这些操作符使用方法的基础上，才可以写出正确的管道聚合程序。
      下一节，将详细介绍聚合管道的代码编写。

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

                                              

      

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