MongoDB聚合管道（AggregationPipeline）

创新互联公司坚持“要么做到，要么别承诺”的工作理念，服务领域包括：成都网站设计、网站建设、企业官网、英文网站、手机端网站、网站推广等服务，满足客户于互联网时代的紫金网站设计、移动媒体设计的需求，帮助企业找到有效的互联网解决方案。努力成为您成熟可靠的网络建设合作伙伴！

管道概念

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以为，还有以下性质的表达式操作符：

组聚合操作符

Bool类型聚合操作符

比较类型聚合操作符

算术类型聚合操作符

字符串类型聚合操作符

日期类型聚合操作符

条件类型聚合操作符

注：以上操作符都必须在管道操作符的表达式内来使用。

各个表达式操作符的具体使用方式参见：

http://docs.mongodb.org/manual/reference/operator/aggregation-group/

聚合管道的优化

   1.$sort  +  $skip  +  $limit顺序优化

如果在执行管道聚合时，如果$sort、$skip、$limit依次出现的话，例如：

{ $sort: { age : -1 } },

{ $skip: 10 },

{ $limit: 5 }

那么实际执行的顺序为：

{ $sort: { age : -1 } },

{ $limit: 15 },

{ $skip: 10 }

$limit会提前到$skip前面去执行。

此时$limit = 优化前$skip+优化前$limit

这样做的好处有两个:1.在经过$limit管道后，管道内的文档数量个数会“提前”减小，这样会节省内存，提高内存利用效率。2.$limit提前后，$sort紧邻$limit这样的话，当进行$sort的时候当得到前“$limit”个文档的时候就会停止。

2.$limit + $skip + $limit + $skip Sequence Optimization

如果聚合管道内反复出现下面的聚合序列：

  { $limit: 100 },

  { $skip: 5 },

  { $limit: 10},

  { $skip: 2 }

首先进行局部优化为：可以按照上面所讲的先将第二个$limit提前：

{ $limit: 100 },

  { $limit: 15},

  { $skip: 5 },

  { $skip: 2 }

进一步优化：两个$limit可以直接取最小值 ，两个$skip可以直接相加:

{ $limit: 15 },

  { $skip: 7 }

3.Projection Optimization

过早的使用$project投影，设置需要使用的字段，去掉不用的字段，可以大大减少内存。除此之外也可以过早使用

我们也应该过早使用$match、$limit、$skip操作符，他们可以提前减少管道内文档数量，减少内存占用，提供聚合效率。

除此之外，$match尽量放到聚合的第一个阶段，如果这样的话$match相当于一个按条件查询的语句，这样的话可以使用索引，加快查询效率。

聚合管道的限制

    1.类型限制

在管道内不能操作 Symbol, MinKey, MaxKey, DBRef, Code, CodeWScope类型的数据( 2.4版本解除了对二进制数据的限制).

     2.结果大小限制

管道线的输出结果不能超过BSON 文档的大小（16M),如果超出的话会产生错误.

     3.内存限制

如果一个管道操作符在执行的过程中所占有的内存超过系统内存容量的10%的时候，会产生一个错误。

当$sort和$group操作符执行的时候，整个输入都会被加载到内存中，如果这些占有内存超过系统内存的%5的时候，会将一个warning记录到日志文件。同样，所占有的内存超过系统内存容量的10%的时候，会产生一个错误。

分片上使用聚合管道

聚合管道支持在已分片的集合上进行聚合操作。当分片集合上进行聚合操纵的时候，聚合管道被分为两成两个部分，分别在mongod实例和mongos上进行操作。

聚合管道使用

首先下载测试数据:http://media.mongodb.org/zips.json 并导入到数据库中。

1.查询各州的人口数

var connectionString = ConfigurationManager.AppSettings["MongodbConnection"];

var client = new MongoClient(connectionString);

var DatabaseName = ConfigurationManager.AppSettings["DatabaseName"];

string collName = ConfigurationManager.AppSettings["collName"];

MongoServer mongoDBConn = client.GetServer();

MongoDatabase db = mongoDBConn.GetDatabase(DatabaseName);

MongoCollection table = db[collName];

var group = new BsonDocument

{

{"$group", new BsonDocument

{

{

"_id","$state"

},

{

"totalPop", new BsonDocument

{

{ "$sum","$pop" }

}

}

}

}

};

var sort = new BsonDocument

{

{"$sort", new BsonDocument{ { "_id",1 }}}

};

var pipeline = new[] { group, sort };

var result = table.Aggregate(pipeline);

var matchingExamples = result.ResultDocuments.Select(x => x.ToDynamic()).ToList();

foreach (var example in matchingExamples)

{

var message = string.Format("{0}- {1}", example["_id"], example["totalPop"]);

Console.WriteLine(message);

}

2.计算每个州平均每个城市打人口数

> db.zipcode.aggregate({$group:{_id:{state:"$state",city:"$city"},pop:{$sum:"$pop"}}},

                              {$group:{_id:"$_id.state",avCityPop:{$avg:"$pop"}}},

                                       {$sort:{_id:1}})

var group1 = new BsonDocument

{

{"$group", new BsonDocument

{

{

"_id",new BsonDocument

{

{"state","$state"},

{"city","$city"}

}

},

{

"pop", new BsonDocument

{

{ "$sum","$pop" }

}

}

}

}

};

var group2 = new BsonDocument

{

{"$group", new BsonDocument

{

{

"_id","$_id.state"

},

{

"avCityPop", new BsonDocument

{

{ "$avg","$pop" }

}

}

}

}

};

var pipeline1 = new[] { group1,group2, sort };

var result1 = table.Aggregate(pipeline1);

var matchingExamples1 = result1.ResultDocuments.Select(x => x.ToDynamic()).ToList();

foreach (var example in matchingExamples1)

{

var message = string.Format("{0}- {1}", example["_id"], example["avCityPop"]);

Console.WriteLine(message);

}

3.计算每个州人口最多和最少的城市名字

>db.zipcode.aggregate({$group:{_id:{state:"$state",city:"$city"},pop:{$sum:"$pop"}}},

                                      {$sort:{pop:1}},

                                      {$group:{_id:"$_id.state",biggestCity:{$last:"$_id.city"},biggestPop:{$last:"$pop"},smallestCity:{$first:"$_id.city"},smallestPop:{$first:"$pop"}}},

                                      {$project:{_id:0,state:"$_id",biggestCity:{name:"$biggestCity",pop:"$biggestPop"},smallestCity:{name:"$smallestCity",pop:"$smallestPop"}}})

var sort1 = new BsonDocument

{

{"$sort", new BsonDocument{ { "pop",1 }}}

};

var group3 = new BsonDocument

{

{

"$group", new BsonDocument

{

{

"_id","$_id.state"

},

{

"biggestCity",new BsonDocument

{

{"$last","$_id.city"}

}

},

{

"biggestPop",new BsonDocument

{

{"$last","$pop"}

}

},

{

"smallestCity",new BsonDocument

{

{"$first","$_id.city"}

}

},

{

"smallestPop",new BsonDocument

{

{"$first","$pop"}

}

}

}

}

};

var project = new BsonDocument

{

{

"$project", new BsonDocument

{

{"_id",0},

{"state","$_id"},

{"biggestCity",new BsonDocument

{

{"name","$biggestCity"},

{"pop","$biggestPop"}

}},

{"smallestCity",new BsonDocument

{

{"name","$smallestCity"},

{"pop","$smallestPop"}

}

}

}

}

};

var pipeline2 = new[] { group1,sort1 ,group3, project };

var result2 = table.Aggregate(pipeline2);

var matchingExamples2 = result2.ResultDocuments.Select(x => x.ToDynamic()).ToList();

foreach (var example in matchingExamples2)

{

Console.WriteLine(example.ToString());

//var message = string.Format("{0}- {1}", example["_id"], example["avCityPop"]);

//Console.WriteLine(message);

}

总结

对于大多数的聚合操作，聚合管道可以提供很好的性能和一致的接口，使用起来比较简单， 和MapReduce一样，它也可以作用于分片集合，但是输出的结果只能保留在一个文档中，要遵守BSON Document大小限制（当前是16M)。

管道对数据的类型和结果的大小会有一些限制，对于一些简单的固定的聚集操作可以使用管道，但是对于一些复杂的、大量数据集的聚合任务还是使用MapReduce。

相关文章：

http://mikaelkoskinen.net/mongodb-aggregation-framework-examples-in-c/