Mongodb Manual阅读笔记:CH6 聚合

  1. 云栖社区>
  2. 博客>
  3. 正文

Mongodb Manual阅读笔记:CH6 聚合

fanr_zh 2014-02-22 22:01:00 浏览587
展开阅读全文

6 聚合

Mongodb Manual阅读笔记:CH2 Mongodb CRUD 操作
Mongodb Manual阅读笔记:CH3 数据模型(Data Models)
Mongodb Manual阅读笔记:CH4 管理
Mongodb Manual阅读笔记:CH5 安全性
Mongodb Manual阅读笔记:CH6 聚合
Mongodb Manual阅读笔记:CH7 索引
Mongodb Manual阅读笔记:CH8 复制集
Mongodb Manual阅读笔记:CH9 Sharding

 

聚合就是,通过处理数据得到一些计算结果。聚合操作可以分组,在Mongodb中聚合可以分为3中:聚合管道(aggregation pipeline),map-reduce,单一目的的聚合方法或者命令

6 聚合... 1

6.1聚合介绍... 2

6.1.1聚合方式... 2

6.1.1.1 聚合管道(aggregation pipeline)2

6.1.1.2 map-reduce. 3

6.1.1.3单一目的的聚合操作... 3

6.1.2其他的特性和能力... 4

6.2聚合概述... 4

6.2.1聚合管道... 5

6.2.1.1 管道... 5

6.2.1.2 管道表达式... 5

6.2.1.3聚合管道特性... 5

6.2.2 Map-Reduce. 5

6.2.2.1 Map-ReduceJavascript函数... 6

6.2.2.2 Map-Reduce的特性... 6

6.2.3单一目的的聚合操作... 6

6.2.3.1 Coutn. 6

6.2.3.2 Distinct. 6

6.2.3.3 Group. 6

6.2.4聚合机制... 6

6.2.4.1聚合管道优化... 7

6.2.4.2聚合管道的限制... 7

6.2.4.3 聚合管道和shard collection. 8

6.2.4.4 Map-Reduceshard集群... 8

6.2.4.5 Map-Reduce的并发... 9

6.3聚合例子... 9

6.3.1邮编的数据集上使用聚合... 9

6.3.1.1数据模型... 9

6.3.1.2返回人口超过100万的州... 9

6.3.1.3 每个州的平均城市人口... 10

6.3.1.4 返回最大和最小城市... 10

6.3.2 用户爱好数据聚合... 10

6.3.2.1数据模型... 10

6.3.2.2 文档排序... 11

6.3.2.3 根据加入的月份排序... 11

6.3.2.4 每月加入的总人数... 11

6.3.2.5前五最受欢迎的爱好... 11

6.3.3 Map-Reduce例子... 12

6.3.3.1 返回每个用户的所有订单价格... 12

6.3.3.2 计算订单平均商品数量... 12

6.3.4 增量执行map-reduce. 13

6.3.4.1 数据... 13

6.3.4.2 初始化执行Map-Reduce. 13

6.3.4.3顺序新增数据... 14

6.3.5 Map函数TroubleShooting. 15

6.3.6Reduce函数Troubleshooting. 15

6.4聚合指南... 15

 

6.1聚合介绍

和查询一样,都是查询collection的文档,不过聚合还要计算,通过计算得到结果

6.1.1聚合方式

6.1.1.1 聚合管道(aggregation pipeline)

聚合管道,通过多个阶段的管道计算,来得到结果。最基本的就是过滤,和查询的过来一样。其他的管道,提供分组的工具,可以计算平均值,sum等。

6.1.1.2 map-reduce

map-reduce分为2个阶段,map阶段,处理每个文档,通过emits发到下个阶段。reduce阶段,输出map阶段的输出组合。map-reduce可以通过finalize阶段处理最后的结果。map-reduce,也可以限制输入条件,也可以排序限制输出结果。

map-reduce的每个阶段都是由javascript脚本写成的函数。javascript提供的很好的灵活性,但是更加复杂和低效。

map-reduce,输出可以超过16MB

6.1.1.3单一目的的聚合操作

单一目的的聚合操作,比如返回行数,distinct等,都来至于同一个collection,但是和map-reduce和聚合管道而言,缺少了灵活性。

6.1.2其他的特性和能力

聚合管道和map-reduce可以在shard集群上使用,map-reduce还可以输出到集群中。

聚合管道,可以通过索引来提高性能。

6.2聚合概述

详细介绍聚合管道,map-reduce,单一目的的聚合操作之间的特性和限制。

6.2.1聚合管道

聚合管道是一个数据聚合模型框架的概念,数据处理管道,通过多个阶段,来计算出聚合结果。

很多情况下由于灵活性,map-reduce是首选但是复杂性很难让人接受。

6.2.1.1 管道

理论上,collection中的文档通过管道来转化计算数据。collection的文档一个一个的通过管道转化,但是并不是每个输入都会有输出。

6.2.1.2 管道表达式

每个管道都有表达式,表达式指定了数据要如何转化。管道只能操作当前文档。一般管道表达式没有当前状态,除了一些特殊的,比如求最大,最小值。

6.2.1.3聚合管道特性

聚合管道把一个collection的数据都丢到管道中转化,这些操作时可以被索引优化的。

管道操作和索引

$match,$sort,$limit,$skip这些操作时可以利用索引的,只要在管道的最开始出,比一下任何操作早就可以了,$project,$unwind,$group

2.4版本中,$geoNear也可以使用地理性的索引,但是必去出现在第一阶段。

简单过滤

若聚合只处理collection的子集,就可以使用一下操作过滤,$match,$limit,$skip,然后放到pipeline的开头可以使用合适的索引。

如果$match放在$sort的后面,那么只有$sort可以使用索引。

其他特性

聚合管道,mongodb提供了内部优化各个管道的顺序,提高执行的性能。

6.2.2 Map-Reduce

map-reduce是处理数据量的范例,在map阶段,应用每个数据,然后搞成k-v对分发出去,在reduce阶段,收集计算聚合数据,然后存放在collection中,最后还要通过finalize函数来输出。

map-reduce内的阶段都是js函数,所以可以在map阶段之前做任何的排序,限制操作。

6.2.2.1 Map-ReduceJavascript函数

使用javascriptmap中形成k-v对,然后在reduce中处理。在现实情况中,map可以又多个key,也可以没有key

6.2.2.2 Map-Reduce的特性

mongodbMap-Reduce的结果可以写入collection也可以直接输出。如果直接输出有BSON的大小限制,16MBMap-Reduce也支持写入到shardcollection中。

6.2.3单一目的的聚合操作

只能用于指定功能。

6.2.3.1 Coutn

返回文档的数量可以是db.collection.count(),也可以是cursor.count()

6.2.3.2 Distinct

查看某个列的distinct值。

db.collection.distinct("cust_id")

6.2.3.3 Group

把查询结果作为输入,通过一些计算然后把结果以数组方式输出。

db.records.group( {

         key: { a: 1 },

         cond: { a: { $lt: 3 } },

         reduce: function(cur, result) { result.count += cur.count },

         initial: { count: 0 }

} )

6.2.4聚合机制

本节介绍,聚合管道的优化,聚合管道的限制,聚合管道和shard集群,map-reduceshard集群,map-reduce并发

6.2.4.1聚合管道优化

聚合管道有个优化阶段,把所有的管道重新排列

管道顺序优化

$sort+$skip+$limit顺序优化:如果你的顺序是$sort,$skip,$limit,那么$limit会被提到$skip上面。如:

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

{ $skip: 10 },

{ $limit: 5 }

优化后

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

{ $limit: 15 }

{ $skip: 10 }

$limit+$skip+$limit+$skip顺序优化:一样还是会把limit放到skip上面,但是limit之间,skip之间会合并,如:

{ $limit: 100 },

{ $skip: 5 },

{ $limit: 10},

{ $skip: 2 }

第一步优化后会把limit放到上面

{ $limit: 100 },

{ $limit: 15},

{ $skip: 5 },

{ $skip: 2 }

第二步优化合并

{ $limit: 15 },

{ $skip: 7 }

porjection优化

如果$project里面指定的是includeMongoDB会把这个projection应用到pipeline的头上。

6.2.4.2聚合管道的限制

类型限制

聚合管道不允许操作一下数据类型:SymbolMinKeyMaxKeyDBRefCodeCodeWScope

结果大小限制

不能操作BSON结果大小16MB

内存限制

聚合使用内存超过内存的10%,过程会报错退出。

一般只有累计操作次才会出现,即要全部输入才会出结果的操作,内存超过5%就会开始告警,并写入日志,超过10%报错退出。

6.2.4.3 聚合管道和shard collection

聚合管道支持在shard collection上运行

shard collection上的聚合操作

第一步,聚合管道会把$group$sort操作发布到每个shard,然后第二个管道再到mongos上运行,这个管道会组合之前的$group,$sort,继续执行。

$groupshard获得子结果,然后组合起来

聚合管道再mongos上的影响

聚合管道会给mongos带来巨大的cpu消耗,如果在shard集群中有大量的聚合管道操作,建议更换体系结构。

6.2.4.4 Map-Reduceshard集群

Map-Reduce输入输出都支持shard

shard集群作为输入

如果shared collection作为数据,mongos会自发的把map-reduce的操作发布到每个shard

shard集群作为输出

如果map-reduce输出带有shard值,mongodb会使用_id作为shard key进行shard

输出到shard collection的条件:

1.如果输出的collection不存在,会根据_id创建shard

2.如果一个空的shard,在map-reduce的第一阶段的结果来是初花shardchunks

3.monogos并发的调度map-reduce的任务到shard中,处理任务的时候,shard会获取自己需要的chunk然后保存

6.2.4.5 Map-Reduce的并发

map-reduce是一个任务组合,包括读输入,执行map,执行reduce,输出

在执行map-reduce的时候会产生以下的锁:

1.读阶段,产生读锁,每100个文档就会释放一次

2.插入到一个零食collection写锁

3.如果collection不存在,创建一个写锁

4.如果collection存在,写入输出写锁

6.3聚合例子

6.3.1邮编的数据集上使用聚合

6.3.1.1数据模型

{

"_id": "10280",

"city": "NEW YORK",

"state": "NY",

"pop": 5574,

"loc": [

-74.016323,

40.710537

]

}

id:邮编,city:城市,state:州,pop:人口,loc:经纬

6.3.1.2返回人口超过100万的州

db.zipcodes.aggregate( { $group :

                                              { _id : "$state",

                                                        totalPop : { $sum : "$pop" } } },

                                     { $match : {totalPop : { $gte : 10*1000*1000 } } } )

1.group,为每个state创建一个文档,sum 人口

2.然后,通过match过滤人口

6.3.1.3 每个州的平均城市人口

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

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

1.groupstatecity人口进行sum

2.state人口进行平均

6.3.1.4 返回最大和最小城市

db.zipcodes.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" } }

         },

         // the following $project is optional, and

         // modifies the output format.

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

                  smallestCity: { name: "$smallestCity", pop: "$smallestPop" } } }

 )

 

1.statecity人口进行sum,并进行排序

2.根据排序获取人口最多,和人口最少的城市

3.通过project修改字段名

6.3.2 用户爱好数据聚合

6.3.2.1数据模型

{

         _id : "jane",

         joined : ISODate("2011-03-02"),

         likes : ["golf", "racquetball"]

}

{

         _id : "joe",

         joined : ISODate("2012-07-02"),

         likes : ["tennis", "golf", "swimming"]

}

6.3.2.2 文档排序

db.users.aggregate(

[

{ $project : { name:{$toUpper:"$_id"} , _id:0 } },

{ $sort : { name : 1 } }

])

6.3.2.3 根据加入的月份排序

db.users.aggregate(

[

         { $project : { month_joined : {$month : "$joined"},name : "$_id",_id : 0},

         { $sort : { month_joined : 1 } }

])

6.3.2.4 每月加入的总人数

db.users.aggregate(

[

         { $project : { month_joined : { $month : "$joined" } } } ,

         { $group : { _id : {month_joined:"$month_joined"} , number : { $sum : 1 } } },

         { $sort : { "_id.month_joined" : 1 } }

])

6.3.2.5前五最受欢迎的爱好

db.users.aggregate(

[

         { $unwind : "$likes" },

         { $group : { _id : "$likes" , number : { $sum : 1 } } },

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

         { $limit : 5 }

])

1.使用unwindlikes进行拆分

2.group对爱和统计,并计数

3.排序

4 5

6.3.3 Map-Reduce例子

Map-Reduce主要针对以下数据模型:

{

         _id: ObjectId("50a8240b927d5d8b5891743c"),

         cust_id: "abc123",

         ord_date: new Date("Oct 04, 2012"),

         status: 'A',

         price: 25,

         items: [      { sku: "mmm", qty: 5, price: 2.5 },

                            { sku: "nnn", qty: 5, price: 2.5 } ]

}

6.3.3.1 返回每个用户的所有订单价格

1.定义map函数,map pricecust_idemits

var mapFunction1 = function() {

                   emit(this.cust_id, this.price);

};

2.定义reduce 函数,2个参数,keycustidvaluesPricesvaluesPrices是一个数组,所以:

var reduceFunction1 = function(keyCustId, valuesPrices) {

         return Array.sum(valuesPrices);

};

3.执行map-reduce,输出指定map_reduce_example如果已经存在会替换原先的collection

db.orders.mapReduce(

         mapFunction1,

         reduceFunction1,

         { out: "map_reduce_example" }

)

6.3.3.2 计算订单平均商品数量

var mapFunction2 = function() {

                            for (var idx = 0; idx < this.items.length; idx++) {

                                     var key = this.items[idx].sku;

                                     var value = {count: 1,qty: this.items[idx].qty};

                            emit(key, value);

}};

var reduceFunction2 = function(keySKU, countObjVals) {

                                               reducedVal = { count: 0, qty: 0 };

                                               for (var idx = 0; idx < countObjVals.length; idx++) {

                                                        reducedVal.count += countObjVals[idx].count;

                                                        reducedVal.qty += countObjVals[idx].qty;

                                               }

                                               return reducedVal;

};

var finalizeFunction2 = function (key, reducedVal) {

         reducedVal.avg = reducedVal.qty/reducedVal.count;

         return reducedVal;

};

db.orders.mapReduce( mapFunction2,reduceFunction2,

                                     {

                                               out: { merge: "map_reduce_example" },

                                               query: { ord_date:{ $gt: new Date('01/01/2012') }},

                                               finalize: finalizeFunction2

                                     }

)

outmerge输出会合并到map_reduce_example

6.3.4 增量执行map-reduce

如果map-reduce的数据集会一直增加,那么不需要对所有的执行map-reduce,最对增加的部分执行即可。

1.在现有的collection上执行map-reduce

2.增加数据,query中指定新数据的条件,在out中指定reduce,合并已有数据

6.3.4.1 数据

db.sessions.save( { userid: "a", ts: ISODate('2011-11-03 14:17:00'), length: 95 } );

db.sessions.save( { userid: "b", ts: ISODate('2011-11-03 14:23:00'), length: 110 } );

db.sessions.save( { userid: "c", ts: ISODate('2011-11-03 15:02:00'), length: 120 } );

db.sessions.save( { userid: "d", ts: ISODate('2011-11-03 16:45:00'), length: 45 } );

db.sessions.save( { userid: "a", ts: ISODate('2011-11-04 11:05:00'), length: 105 } );

db.sessions.save( { userid: "b", ts: ISODate('2011-11-04 13:14:00'), length: 120 } );

db.sessions.save( { userid: "c", ts: ISODate('2011-11-04 17:00:00'), length: 130 } );

db.sessions.save( { userid: "d", ts: ISODate('2011-11-04 15:37:00'), length: 65 } );

6.3.4.2 初始化执行Map-Reduce

var mapFunction = function() {

                                     var key = this.userid;

                                     var value = {userid: this.userid,

                                                        total_time: this.length,

                                                        count: 1,

                                                        avg_time: 0

                                     };

                                     emit( key, value );

};

var reduceFunction = function(key, values) {

                                                        var reducedObject = {

                                                                 userid: key,

                                                                 total_time: 0,

                                                                 count:0,

                                                                 avg_time:0

                                                        };

                                                        values.forEach( function(value) {

                                                                 reducedObject.total_time += value.total_time;

                                                                 reducedObject.count += value.count;

                                                        });

                                                        return reducedObject;

};

var finalizeFunction = function (key, reducedValue) {

                                                        if (reducedValue.count > 0)

                                                        reducedValue.avg_time = reducedValue.total_time / reducedValue.count;

                                                        return reducedValue;

};

db.sessions.mapReduce( mapFunction,

                                               reduceFunction,

                                               {

                                                        out: { reduce: "session_stat" },

                                                        finalize: finalizeFunction

                                               }

)

6.3.4.3顺序新增数据

db.sessions.save( { userid: "a", ts: ISODate('2011-11-05 14:17:00'), length: 100 } );

db.sessions.save( { userid: "b", ts: ISODate('2011-11-05 14:23:00'), length: 115 } );

db.sessions.save( { userid: "c", ts: ISODate('2011-11-05 15:02:00'), length: 125 } );

db.sessions.save( { userid: "d", ts: ISODate('2011-11-05 16:45:00'), length: 55 } );

 

db.sessions.mapReduce( mapFunction,

                                     reduceFunction,

                                     {

                                               query: { ts: { $gt: ISODate('2011-11-05 00:00:00') } },

                                               out: { reduce: "session_stat" },

                                               finalize: finalizeFunction

                                     }

);

6.3.5 Map函数TroubleShooting

1.定义map函数

var map = function() {

emit(this.cust_id, this.price);

};

2.定义reduce函数

var emit = function(key, value) {

print("emit");

print("key: " + key + " value: " + tojson(value));

}

3.调用map函数

var myDoc = db.orders.findOne( { _id: ObjectId("50a8240b927d5d8b5891743c") } );

map.apply(myDoc);

4.验证输出

emit

key: abc123 value:250

5.多文档调用

var myCursor = db.orders.find( { cust_id: "abc123" } );

while (myCursor.hasNext()) {

         var doc = myCursor.next();

         print ("document _id= " + tojson(doc._id));

         map.apply(doc);

         print();

}

6.验证结果

6.3.6Reduce函数Troubleshooting

1.确定返回的数据类型,和传入的数据类型一致

2.数组的顺序,不影响结果

3.多次执行不影响结果

6.4聚合指南

查看手册p308

网友评论

登录后评论
0/500
评论
fanr_zh
+ 关注