MongoDB 聚合管道（Aggregation Pipeline）

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了MongoDB 聚合管道（Aggregation Pipeline），小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含13580字，纯文字阅读大概需要20分钟。

内容图文

管道概念

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

与这种方式设计相似的例如netty 的 ChannelPipeiline，还是更加广泛的Servlet Fileter过滤器，这类拦截器实际上是职责链方式的一种变形，主要是为了方便时间的拦截和用户逻辑的定制。

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

以netty为例子，下图中Channel为数据管道，ChannelPipeline为ChannelHandler的容器，ChannelHandler既可以是netty预先定义的handler处理，也可以是自定义的handler处理方式，当event通过管道进入到ChannelPipeline中，ChannelPipeline会通过预先定义好顺序的handler对源数据进行分部处理，每经过一道handler，数据被处理成为了另一种形态或变换，每个handler处理的内容不同，handler的输入为上一个handler的输出，输出成为下一个handler的输入。

回到mongodb的管道操作，它的基本功能有两个：

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

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

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

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

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

管道操作符的种类：

管道操作符详细使用说明

  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会返回一些坐标值，这些值以按照距离指定点距离由近到远进行排序

管道表达式

管道操作符作为“键”,所对应的“值”叫做管道表达式。例如上面例子中{$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上进行操作。

总结

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

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

转载来源：https://www.cnblogs.com/shanyou/p/3494854.html

内容总结

以上是互联网集市为您收集整理的MongoDB 聚合管道（Aggregation Pipeline）全部内容，希望文章能够帮你解决MongoDB 聚合管道（Aggregation Pipeline）所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端