mongodb 聚合管道

管道概念

先介绍管道的概念，在POSIX多线程的使用方式中，定义了一种重要的pipeline方式，成为“流水线”或“管道”，这种方式使得数据被一组线程顺序执行，其流程以下：

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

mongodb中的管道

mongodb在2.2版本中引入了聚合框架（aggregate framework）的新功能，它是聚合的新框架，其概念相似于数据处理的管道，每一个文档通过一个由多个节点组成的管道，每一个节点至关于流水线中的一个stage，有本身的功能（分组，过滤等），文档通过管道处理后，最后输出相应的结果，管道的基本功能有两个：

1. 对文档进行“过滤”，筛选出合适的文档
2. 对文档进行“变换”，改变文档的输出形式

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

mongodb中的聚合（aggregate）主要用于简单的数据处理（平均值，求和等），并返回计算后的数据结果，相似于sql中的内嵌函数（count（）等） 在mongodb官网给出了聚合框架的应用实例：

能够看到，一个聚合管道中包含多个stage，每一个stage都是对数据的一次处理，mongodb中的聚合采用aggregate()方法，语法以下： >db.COLLECTION_NAME.aggregate(pipeline,options) 其中pipeline为一个array，语法为： [{<stage1>,<stage2>,<stage3>,...}] 其中每一个的语法为： {$管道操做符：{ 管道表达式 } }

那么下面列举一些较常见的管道操做符以及他们的做用，后文还会继续给出实例：

操做符	描述	语法
$project	数据投影，主要用于重命名，增长，删除字段	db.article.aggregate({ $project : {title : 1 ,author : 1 ,}});
$match	过滤，筛选符合条件的文档，做为下一阶段输入	db.articles.aggregate( [{ $match : { score : { $gt : 70, $lte : 90 } } },{ $group: { _id: null, count: { $sum: 1 } } }] );
$limit	限制通过管道的文档数量	db.article.aggregate({ $limit : 5 });
$skip	待操做集合处理前跳过部分文档	db.article.aggregate({ $skip : 5 });
$unwind	将数组拆分红独立字段	db.article.aggregate({$project:{author:1,title:1,tags:1}},{$unwind:"$tags"})
$group	对数据进行分组	db.article.aggregate({ $group : {_id : "$author",docsPerAuthor : { $sum : 1 },viewsPerAuthor : { $sum : "$pageViews" }}});
$sort	对文档按照指定字段排序	db.users.aggregate( { $sort : { age : -1, posts: 1 } });
$sample	随机选择从其输入指定数量的文档。	{ $sample: { size: <positive integer> } }
$out	必须为pipeline最后一个阶段管道，由于是将最后计算结果写入到指定的collection中
$indexStats	返回数据集合的每一个索引的使用状况	{ $indexStats: { } }

更多的管道操做符参考mongodb官方文档，官方文档写的更为详细，可是对语法的描写较少 docs.mongoing.com/manual-zh/m…

管道操做符示例

下面针对经常使用管道操做符举一些例子：

先加载数据：

use test1

db.mycol.remove({})

document1=({name:'dogOne',age:1,tags:['animal','dog'],type:'dog',money:[{min:100},{norm:200},{big:300}]});

document2=({name:'catOne',age:3,tags:['animal','cat'],type:'cat',money:[{min:50},{norm:100},{big:200}]});

document3=({name:'catTwo',age:2,tags:['animal','cat'],type:'cat',money:[{min:20},{norm:50},{big:100}]});

document4=({name:'dogTwo',age:5,tags:['animal','dog'],type:'dog',money:[{min:300},{norm:500},{big:700}]});

document5=({name:'appleOne',age:0,tags:['fruit','apple'],type:'apple',money:[{min:10},{norm:12},{big:13}]});

document6=({name:'appleTwo',age:0,tags:['fruit','apple'],type:'apple',money:[{min:10},{norm:12},{big:13}]});

document7=({name:'pineapple',age:0,tags:['fruit','pineapple'],type:'pineapple',money:[{min:8},{norm:9},{big:10}]});

db.mycol.insert(document1)

db.mycol.insert(document2)

db.mycol.insert(document3)

db.mycol.insert(document4)

db.mycol.insert(document5)

db.mycol.insert(document6)

db.mycol.insert(document7)
复制代码

下面是执行结果：

/* 1 */

{

    "_id" : ObjectId("59187984f322c585a98664e2"),

    "name" : "dogOne",

    "age" : 1.0,

    "tags" : [ 

        "animal", 

        "dog"

    ],

    "type" : "dog",

    "money" : [ 

        {

            "min" : 100.0

        }, 

        {

            "norm" : 200.0

        }, 

        {

            "big" : 300.0

        }

    ]

}



/* 2 */

{

    "_id" : ObjectId("59187984f322c585a98664e3"),

    "name" : "catOne",

    "age" : 3.0,

    "tags" : [ 

        "animal", 

        "cat"

    ],

    "type" : "cat",

    "money" : [ 

        {

            "min" : 50.0

        }, 

        {

            "norm" : 100.0

        }, 

        {

            "big" : 200.0

        }

    ]

}



/* 3 */

{

    "_id" : ObjectId("59187984f322c585a98664e4"),

    "name" : "catTwo",

    "age" : 2.0,

    "tags" : [ 

        "animal", 

        "cat"

    ],

    "type" : "cat",

    "money" : [ 

        {

            "min" : 20.0

        }, 

        {

            "norm" : 50.0

        }, 

        {

            "big" : 100.0

        }

    ]

}



/* 4 */

{

    "_id" : ObjectId("59187984f322c585a98664e5"),

    "name" : "dogTwo",

    "age" : 5.0,

    "tags" : [ 

        "animal", 

        "dog"

    ],

    "type" : "dog",

    "money" : [ 

        {

            "min" : 300.0

        }, 

        {

            "norm" : 500.0

        }, 

        {

            "big" : 700.0

        }

    ]

}



/* 5 */

{

    "_id" : ObjectId("59187984f322c585a98664e6"),

    "name" : "appleOne",

    "age" : 0.0,

    "tags" : [ 

        "fruit", 

        "apple"

    ],

    "type" : "apple",

    "money" : [ 

        {

            "min" : 10.0

        }, 

        {

            "norm" : 12.0

        }, 

        {

            "big" : 13.0

        }

    ]

}



/* 6 */

{

    "_id" : ObjectId("59187984f322c585a98664e7"),

    "name" : "appleTwo",

    "age" : 0.0,

    "tags" : [ 

        "fruit", 

        "apple"

    ],

    "type" : "apple",

    "money" : [ 

        {

            "min" : 10.0

        }, 

        {

            "norm" : 12.0

        }, 

        {

            "big" : 13.0

        }

    ]

}



/* 7 */

{

    "_id" : ObjectId("59187984f322c585a98664e8"),

    "name" : "pineapple",

    "age" : 0.0,

    "tags" : [ 

        "fruit", 

        "pineapple"

    ],

    "type" : "pineapple",

    "money" : [ 

        {

            "min" : 8.0

        }, 

        {

            "norm" : 9.0

        }, 

        {

            "big" : 10.0

        }

    ]

}
复制代码

1.$project操做符与$match操做符

$project管道操做符用于修改流中的文档，$match管道操做符用于对流中的文档进行过滤，仅容许符合条件的文档进入下一个阶段，过滤操做不会修改文档。$match操做使用mongodb标准的查询条件，对于每个输入文档，若是符合条件，则输出这个文档，不然丢弃该文档。因为aggregate管道对于内存的限制，在处理大文件的时候，最好先用match操做符进行筛选，减小内存占用。

假定咱们想提取money中min为100的文档，而且只输出名称和money数组中的min那一项，用$project与$match操做符能够很好的实现

use test1
db.mycol.aggregate(
    {$match:{'money.min':100}},
    {$project:{_id:0,name:'$name',minprice:'$money.min'}}
    )
复制代码

输出结果为：

/* 1 */

{

    "name" : "dogOne",

    "minprice" : [ 

        100.0

    ]

}
复制代码

能够发现，在project操做符后，文档中的字段被改变了。 也要注意到，对于数组中对象的引用，须要采用 '$money.min'形式 注意：

1. 不能在$match操做符中使用$where表达式操做符。
2. $match尽可能出如今管道的前面，这样能够提前过滤文档，加快聚合速度。
3. 若是$match出如今最前面的话，可使用索引来加快查询。

2.$limit $skip操做符

$limit与$skip操做符是用于限制与跳过相应文档，与find中的limit与skip方法效果相同。 假定咱们想提取money中min小于100的文档，而且限制3个文档，跳过一个文档再显示 脚本为

use test1

db.mycol.aggregate(

    {$match:{'money.min':{$lt:100}}},

    {$limit:3},

    {$skip:1},

    {$project:{_id:0,name:'$name',minprice:'$money.min'}}

    )
复制代码

结果为：

/* 1 */

{

    "name" : "catTwo",

    "minprice" : [ 

        20.0

    ]

}



/* 2 */

{

    "name" : "appleOne",

    "minprice" : [ 

        10.0

    ]

}
复制代码

能够发现结果知足咱们的需求

3.$group操做符

$group操做符用来对数据进行分组。 $group的时候必需要指定一个_id域，同时也能够包含一些算术类型的表达式操做符 好比咱们要经过type类型来对数据进行分类，而且同时统计他们的年龄age总和， 脚本为：

use test1

db.mycol.aggregate(

    {$group:{_id:'$type',sumage:{$sum:'$age'}}}

)
复制代码

结果为：

/* 1 */

{

    "_id" : "pineapple",

    "sumage" : 0.0

}



/* 2 */

{

    "_id" : "cat",

    "sumage" : 5.0

}



/* 3 */

{

    "_id" : "apple",

    "sumage" : 0.0

}



/* 4 */

{

    "_id" : "dog",

    "sumage" : 6.0

}
复制代码

能够看到数据按照 猫，狗，苹果，菠萝进行了分类，而且年龄相加了。 注意：

1. $group的输出是无序的。
2. $group操做目前是在内存中进行的，因此不能用它来对大量个数的文档进行分组。
3. 必须指定 _id 的域

4. $sort操做符

sort操做符用来对数据进行排序，一样1表明升序，-1表明降序 假定咱们按照年龄对数据进行排序，为了减小输出行数，咱们用上分组与skip 脚本为：

use test1

db.mycol.aggregate(

    {$group:{_id:'$type',sumage:{$sum:'$age'}}},

    {$skip:1},

    {$sort:{sumage:1}}

)
复制代码

结果为：

/* 1 */

{

    "_id" : "apple",

    "sumage" : 0.0

}



/* 2 */

{

    "_id" : "cat",

    "sumage" : 5.0

}



/* 3 */

{

    "_id" : "dog",

    "sumage" : 6.0

}
复制代码

注意：

1. 若是将$sort放到管道前面的话能够利用索引，提升效率
2. MongoDB 对内存作了优化，在管道中若是$sort出如今$limit以前的话，$sort只会对前 $limit个文档进行操做，这样在内存中也只会保留前$limit个文档，从而能够极大的节省内存
3. $sort操做是在内存中进行的，若是其占有的内存超过物理内存的10%，程序会产生错误

最后

其余不经常使用的操做符暂不说明。 须要额外注意操做符对内存的要求。