Mongodb是目前最受欢迎的大数据存储平台之一，它可以作为云计算技术的底层存储层，比如为spark、hadoop、pig、hive、drill等计算框架提供源数据。Mongodb本身也提供了aggregation、mapreduce特性，以支持对大数据的计算、统计、分类等需求。
Aggregation简单来说，就是提供数据统计、分析、分类的方法，这与mapreduce有异曲同工之处，只不过mongodb做了更多的封装与优化，让数据操作更加便捷和易用。Aggregation操作，接收指定collection的数据集，通过计算后返回result数据；一个aggregation操作，从input源数据到output结果数据，中间会依次经过多个stages，整体而言就是一个pipeline；目前有10种stages，我们稍后介绍；它还提供了丰富的Expression（表达式）来辅助计算。我们展示一个例子：
SQL语义：
select zipcode, count(*) AS total from <collection> where status = 0 group by zipcode order by total
原始数据结构：
{ "zipcode" : 1000000, "amount" : 5, "created" : { "$date" : 1448445890763 }, "status" : 1 }
Aggregation查询语句：
db.<collection>.aggregate([
{"$match" : { "status" : 0 }},
{"$group" : { "_id" : "$zipcode", "total" : { "$sum" : "$amount" }}},
{"$sort":{total : 1}},
{"$out":"_group_"}])
JAVA代码示例：
List<Document> pipeline = new ArrayList<Document>();
Document match = new Document("$match",new Document("status",0));
pipeline.add(match);//match stage
Document group = new Document();
group.put("$group", new Document("_id", "$zipcode").append("total", new Document("$sum", "$amount")));
pipeline.add(group);
Document sort = new Document("$sort",new Document("total",1));
pipeline.add(sort);
String output = "_group_";
Document out = new Document("$out",output);
pipeline.add(out);
source.aggregate(pipeline)
.allowDiskUse(true);
MongoCollection<Document> result = db.getCollection(output);
MongoCursor<Document> cursor = result.find().batchSize(100).iterator();
while (cursor.hasNext()) {
Document item = cursor.next();
System.out.println(item.toJson());
}
cursor.close();
此aggregation操作最终的结果输出到了“_group_”临时表中，我们可以查看它的结果数据大致如下：
{ "_id" : 1000005, "total" : 43425 }
{ "_id" : 1000008, "total" : 43809 }
{ "_id" : 1000003, "total" : 44716 }
{ "_id" : 1000000, "total" : 45130 }
{ "_id" : 1000002, "total" : 45205 }
...
mongodb已经提供了三个“单一用途的”aggregation操作：
1）count：计算符合query的documents的总个数，语义同“select count(*) from <table> where <query>”，mongodb的语法为：
db.<collection>.count({"status":0})
2）distinct：获取符合query中指定filed的不同值的列表，语义同“select distinct(<field>) from <table> where <query>”。
db.<collection>.distinct("zipcode",{"status" : 0})
3）group：分组，注意它和aggregation中的“$group”有些类似，但不是一个操作，因为group无法在sharding环境中使用，所以mongodb建议使用“$group”进行替代，此处不再详解，请参见【group操作】
一、特性
1、Aggregation有几个核心的特性：
1）支持多种stages
2）可以将计算结果保存在collection中，在sharding环境中仍然适用，而且在output之前可以对结果数据进行“修剪”；当然可以将结果数据保存在内存（inline）并返回cursor，便于客户端访问结果数据。
3）同一种stage可以出现多次 ，这相对于mapreduce而言是个巨大的进步。（这种情况通常需要多个mapreduce才能完成）
4）stages中可以使用内置的多种计算操作、表达式，几乎可以覆盖绝大部分数据统计、分析的需求，所以用aggregation开发日常数据处理任务，是非常简单的事情。当然相对于mapreduce这种更加开放的计算模型（基于javascript语法和解析引擎），aggregation在计算复杂的数据任务时还不够灵活，但是书写mapreduce脚本也是一件复杂的事情。
2、局限性
无论如何，aggregation返回的计算结果，每条document的大小不得超过16M（mongodb中document的最大尺寸，此参数不可修改），在下文中我们会了解到$push、$addToSet等计算操作，有可能会导致此问题的发生。在一些计算方式中，可能最终结果只产出一条document，那么就更需要注意此问题。通常情况下，aggregation可以output任意多条documents，结果数据的总size是没有限制的，但是结果不能输出到sharding collection中，这与mapreduce是有区别的。
如果你从事过大数据计算、数据统计等相关工作，应该知道每个计算任务（job或task）都会使用独立的有限大小的内存空间，mongodb没有提供复杂的内存分配模型（任务调度算法），只是简单的限定每个stage最多使用100M内存，如果超过此值将终止计算并返回error；为了支持较大数据集合的处理，我们可以指定“allowDiskUse”参数将“溢出”的数据写入本地的临时文件中（临时的collection），这个参数我们通常需要设定为true。（参见上述示例）
3、优化
aggregation操作，简单而言就是将整个collection的数据全部input到pipeline中，经过多个stages计算之后，输出少量数据的过程。由mongodb引擎的特性决定，读取整个collection数据对磁盘和内存消耗都是非常巨大的，即使是在sharding环境中；那么尽可能的减少input数据量可以有效的提升处理的效率，对于有sort操作的，就需要更加谨慎，因为排序更加消耗内存和CPU。那么提高效率的手段之一，就是尽可能的使用索引（indexes），让$match和$sort尽可能的出现在pipeline的开端，$match使用索引可以更加高效的过滤input数据集进而不需要全表扫描，$sort使用索引可以避免额外的排序（因为索引本身就是有序的），索引只会在访问原始document时有效，如果是pipeline的中间数据，比如$group之后的$sort将无法使用到索引；当然索引的建立，是依据application中数据的访问模式而定，我们通常不会根据aggregation的需要而增加索引。
如果只需要对部分数据进行计算，那么可以将$match、$limit、$skip放置在pipeline的开端，那么handler只需要读取有效的、较少数量的documents；不恰当的做法，是将$limit、$skip放置在pipeline的末端，这意味着可能读取、处理更多的数据，产生更多的结果集，这意味着对性能的浪费。
4、sharded collection
上文已经提到sharding collection不支持group方法（即db.<collection>.group()，而非pipeline中的$group）；但是它支持aggregation pipeline和mapreduce操作。当aggregation操作sharded collection时，pipeline将会被分成2部分：源数据过滤（filters），和后期处理（prost-processing）；第一部分主要就是从collection中读取数据，或者根据索引过滤数据，位于pipeline开端的$match、$limit将会包含在内，mongos将会把aggregation操作发给特定的shards，如果$match中不包含sharding key，那么将会把操作发给所有的shards，并在每个shard上执行aggregation的第一部分的stages，此部分最大的特点就是不对数据进行计算，仅仅筛选并创建cursor；第二部分是后期处理部分，它将在primary shard上运行（概念参见【primary shard】），通常为$group、$unwind、$out、$project，其他stages也可以在第二部分，priamry合并其他shards上的cursor（迭代读取数据），读取数据并执行aggregation的第二部分的操作，然后将最终执行结果返回给mongos。（旧版中第二部分pipeline在mongos上执行）；由此可见，尽可能早的过滤数据，可以有效降低primary的负载压力。我们可以在aggregate方法中指定“explain”参数，查看各个阶段执行的计划，其中cursor字段即为第一部分，其他的为第二部分，我们也能看出cursor分布在shards的情况。
db.orders.aggregate(
[
{ $match: { status: "A" } },
{ $group: { _id: "$cust_id", total: { $sum: "$amount" } } },
{ $sort: { total: -1 } }
],
{
explain: true
}
)
其中limit、skip在各个shards上的分发原理，请参见【mongos部分】
3.2版本调整：mongodb为了提升sharding集群的整体性能，除了$out和$lookup需要在primary shard上运行外，其他的stages均可以并行的在符合匹配规则的shards执行，然后将各自的results数据转发给其中某个shard做最后的merge，这种方式有效了减轻了primary shard的负载压力；简单而言，3.2的pipleline处理模式基本没有改变，只是“第二部分”工作将从primary shard上转义到其中某个shard上。
二、Stages
Pipeline是有多个stages组成，每个stage提供不同的数据操作，stages在声明时是数组结构，具有有序性，当前stage的计算结果将作为下一个stage的输入，除$out之外，每种stages均可以在pipeline中出现多次，aggregate支持如下几种stages：
1、$project
与mongodb查询语句中的“projection”功能类似，对输入的document字段进行“取舍”，指定哪些字段将会包含（include）或者不包含（exclude）在输出中。
$project:{_id:0,name:{$concat:["$last", "," ,"$first"},"userid":1}
默认情况下“_id”字段一定被包含，除非显式指定“_id : 0”才能排除掉；“0”或者false表示“不包含”，“1”或者true表示包含，如果值是一个expression，则表示新增或者重置字段的值。比如上述文档，表示_id字段排除掉，“userid”字段包含，增加一个新字段“name”，其值为“last”和“first”两个字段值的拼接。
当需要增加一个字段，它的值确实是1、0、true或者false时，这就与$project语义有些冲突，为了确保值原样输出，那么就需要告知$project不要解析它们，这需要使用$literal操作符：
$project:{"sum":{$literal : 1}}
和document访问模式一样，$project也支持通过“.”路径访问内嵌文档：
$project:{_id:0,"friends.name":1}或者
$project:{_id:0,"firends":{"name":1}}
3.2+版本之后，$avg、$max、$min、$sum这四个“聚合函数”（原来只能在$group中使用）也可以在$project中使用了。
2、$match
根据query条件筛选文件，符合条件的文件将会传递给下一个stage。$match的语法和mongodb query语法一样，且$match中不能使用aggregate expression或者比较操作，只能使用query操作允许的操作符（参见Query操作）。$match用于筛选数据，为了提高后续的数据处理效率，尽可能的将$match放于pipeline的前端以减少数据读取或者计算量；当然$match也可以放在$out之前用于控制数据输出量。如果$match放于pipeline的开始，还可以使用到indexes机制提高数据查询的性能。
$match:{"status":1}
3、$redact
根据字段所处的document结构的级别，对文档进行“修剪”，它通常和“判断语句if-else”结合使用即“$cond”。$redact可选值有3个：
1）$$DESCEND：包含当前document级别的所有fields。当前级别字段的内嵌文档将会被继续检测。
2）$$PRUNE：不包含当前文档或者内嵌文档级别的所有字段，不会继续检测此级别的其他字段，即使这些字段的内嵌文档持有相同的访问级别。
3）$$KEEP：包含当前文档或内嵌文档级别的所有字段，不再继续检测此级别的其他字段，即使这些字段的内嵌文档中持有不同的访问级别。
{
_id: 1,
tags: [ "G", "STLW" ],
year: 2014,
subsections: [
{
subtitle: "Section 1",
tags: [ "SI", "G" ],
},
{
subtitle: "Section 2",
tags: [ "STLW" ],
},
{
subtitle: "Section 3",
tags: [ "TK" ],
content: {
tags: [ "HCS" ]
}
}
]
}
那么对于语句：
$redact: {$cond: {
if: {
$gt: [ { $size: { $setIntersection: [ "$tags", ["STLW","G"] ] } }, 0 ] },
then: "$$DESCEND",
else: "$$PRUNE"
}
}
$setIntersection表示将2个数组的交集中不同元素的个数，$cond就是一个三元表达式，此例中表示“如果交集元素的个数大于0，则值为为$$DESCEND，否则为$$PRUNE”。对于此文档（ROOT级别）的最高级别的tags值为["G","STLW"]，此级别值为$$DESCEND，即此tage同级别的其他字段将会包含；那么继续检测“subsections.tags”级别的所有文档（是个数组，则逐个检测），基本思路类似，如果此级别返回$$DECEND那么继续检测“subsections.tags.content.tags”是否符合访问规则，如果返回$$PRUNE，那么此tags所在的内嵌文档的所有字段将被排除，即使与此tags同级别的contents.tags符合访问规则。最终输出结果：
{
"_id" : 1,
"tags" : [ "G", "STLW" ],
"year" : 2014,
"subsections" : [
{
"subtitle" : "Section 1",
"tags" : [ "SI", "G" ],
"content" : "Section 1"
},
{
"subtitle" : "Section 2: Analysis",
"tags" : [ "STLW" ],
"content" : "Section 2"
}
]
}
4、$limit、$skip、$sort
同Query中的limit、skip、$sort限定符。如果$limit、$sort相邻出现，那么mongodb将会使用“TOP N”计算方式来处理数据，如果N较小这些数据可以在内存中排序完成，如果N很大（> 100M），则需要指定“allowDiskUse”为true，否则会发生错误。
5、$unwind
将指定的数组结构拆解成多条document，其中指定的Field必须是数组。3.2+版本之后，Field可以不是数组，此时将会把它当做只有一个item的数组处理。比如文档：
{"_id" : 1,"item":"ABC1","sizes":["S","M","L"]}
执行$unwind:"$sizes"时输出结果为：
{ "_id" : 1, "item" : "ABC1", "sizes" : "S" }
{ "_id" : 1, "item" : "ABC1", "sizes" : "M" }
{ "_id" : 1, "item" : "ABC1", "sizes" : "L" }
6、$out
必须为pipeline最后一个stage，将计算结果写入到指定的collection中，格式：$out : <collection>，不支持sharded collection，也不支持capped collection。如果指定的collection不存在，则直接创建。如果此collection已经存在，那么结果数据将原子性的替换旧的collection，即在结果数据写入时，不会改变原collection数据和索引，当数据写入完毕后原子性替换，，如果在写入出错时，原collection的数据不会有任何改变。
7、$group
分组，语义同SQL中的“group by”，语法：$group:{_id:<group key>，<field>:{<accumulator>:<expression>},...}。“_id”字段是必须的，其值是group操作的key，可以为null或者计算表达式；可以增加其他自定义的字段，字段值可以为group支持的计算表达式（Expressions），$group首先将数据根据key进行分组，然后再对每个组执行如下的表达式计算：
1）$sum：对每个group指定字段值进行累加计算。忽略非数字的值。
$group:{_id:"$productId","totalAmount":{$sum : "$quantity"}}
##在order表中，统计每个商品的销售总数量
2）$avg：对每个group进行“平均值”，忽略非数字的值。
$group:{_id:"$categoryId","avgPrice":{$avg : "$price"}}
3）$first：返回每个group的第一条数据，顺序有$sort决定，如果没有排序，默认为文档的自然存储顺序。
$sort:{categoryId : 1,price : 1},
$group:{_id:"categoryId","cheapest":{$first:"$price"}}
4）$last：返回每个group的最后一条数据，顺序有$sort决定，如果没有排序，默认为文档的自然存储顺序。
5）$max、$min：获取每个group中最大、最小值。
$sort:{categoryId:1}
$group:{_id:"categoryId","maxPrice":{$max:"$price"}}
##获取每个品类下，价格最高的商品
6）$push：将指定的表达式的值添加到一个数组中，注意最终返回的document尺寸不要超过16M。
$group:{_id:"author","bookIds":{$push:"$bookId"}}
##获取每个作者的图书ID的列表
7）$addToSet：将表达式的值添加到一个集合中（无重复值），注意最终分那会的document尺寸不要超过16M。
$group:{_id:"categoryId","shopIds":{$addToSet:"$shopId"}}
##获取每个品类下，不同商铺的ID
上述7个操作符只能在$group中使用，$group通常用于获取某一维度不同的值、计算总和、分类数据，是aggregation操作中最重要的stage；如果将group的功能转换成mapreduce可能需要书写大量的代码。$group的key可以为null或者一个常量值，用于计算全局的数据，比如统计所有“商品的库存总价值”：
$group:null,"total":{$sum:{$multiply:["$price","$quantity"]}}
8）$geoNear：用于地理位置数据分析，参见【geoNear】
9）$sample：
3.2+版本中新增，类似于大数据中的“数据采样”方式，从input中随机选择N条documents，语法“{$sample: {size: <N>}}”。如果N大于collection中总数据的5%，那么$sample将会执行collection扫描、sort，然后选择top N条文档；如果N小于5%，对于wiredTiger而言则会遍历collection并使用“伪随机”的方式选取N条文档，对于MMAPv1引擎则在_id索引上随机选取N条文档。（本文没有测试$sample）
10）$lookup
3.2+版本新增，“left outer join”功能；从事过大数据开发的人都知道，如果仅仅基于mapreducer计算模式，实现跨文件“join”还是一件比较复杂的事情；mongodb提供了$lookup以支持简单的join，以当前collection作为主表（左端），将另一个non-sharded collection作为辅表参与join。语法：
{
$lookup:
{
from: <参与join的辅表>,
localField: <参与join匹配的本表字段>
foreignField: <参与join的辅表字段>
as: <将辅表数据输出到此字段中>
}
}
其中from表示一个参与join的辅表，这个表不能是sharding collection；localField和foreignField就是参与join的key，如果documents不存在上述field，那么将以null参与匹配；“as”指定一个输出字段，将辅表匹配成功的documents添加到此“字段”中，作为结果输出。
##比如主表为orders，数据结构为：
{_id:1,"product_id":1000,"price":30.0}
##辅表为products，数据结果为：
{_id:1,"id":1000,"name":"mongodb reference"}
{
$lookup: {
from:"products",
localField:"product_id",
foreignField:"id",
as:"products"
}
}
执行结果可能类似于：
{
_id:1,
"product_id":1000,
"price":32.0,
"products":[
{_id:1,id:1000,"name":"mongodb reference"}
]
}
三、Expressions（表达式）
在$group中我们已经了解到了几个expressions，此外aggregation还支持多种其他类型的表达式，它们为计算过程提供了很大的便利性。$match只能使用query支持的表达式，$group中特定的几个表达式只能在$group中使用，对于其他的stage均可以使用下文中介绍的表达式。【参见表达式】
表达式中可以使用“field path”（字段路径）来访问输入文档的字段值，字段路径有“$”和字段名称构成，即“$<field>”，比如：$userid表示访问useid字段值，支持内嵌文档的访问，比如$user.name。有一个特殊的字段路径，即$$ROOT或者$$CURRENT，表示当前文档，$$ROOT.userid其实和$userid一样，我们通常在使用字段路径时忽略$$ROOT。$$ROOT、$$CURRENT和上文提到的$$DESCEND、$$PRUNE、$$KEEP一样，同属系统变量。不过通常我们还需要在aggregate中声明自定义的变量。
1、自定义变量
1）$let
指定临时变量，并返回变量计算的结果。$let包含2个字段：vars和in，其中vars用该声明临时变量，in用来计算，其中vars中的变量只能在in中使用，其生命周期也只在当前$let中有效；为了避免与“字段路径”的访问模式混淆，在in中访问vars中的变量需要使用$$前缀。示例：
$project:{
totalAmount :{
$let {
vars : {
total : {$multiply : {$price,$quantity}},
in: {$subtract : {'$$total','$discounted'}}
}
}
}
在$let中声明一个变量total，值为价格与数量乘积，最终totalAmount的值为“总价 - 折扣”。
2）$map
对array数组元素进行迭代，并返回一个新的数组。类似一个foreach过程，语法结构如下：
$map : {input:<expression>,as:<item>,in:<expression>}
它有三个字段，input表示需要计算的数组，as表示当前数组元素的别名，in可以使用当前元素的别名进行计算；示例如下：
##原始数据
{_id : 1, "numbers" : [2,5,8]}
{_id : 2, "numbers" : [0,1,3]}
##aggregation
$project : {
"numbers" : {
$map : {input : "$numbers",as:"item",in:{$add : {'$item',2}}}
}
}
##result
{_id : 1, "numbers" : [4,7,9]}
{_id : 2, "numbers" : [2,3,5]}
2、$literal（字面值）
或许大家都比较熟悉这个概念，即$literal的值仅作字符串处理，原样输出，不进行表达式解析和计算，这类似于XML结构中的CDATA。比如：
$literal : {$add: [2,3]} 最终输出"$add:[2,3]"
$literal : "$name" 最终输出"$name"
##----示例----
##原始文档
{_id:1,"product":"book1","price":"$10"}
{_id:1,"product":"book2","price":"45.01"}
##aggregation
{$project:
{costByDollar:{
$eq : [
{$literal : "$"},{$substr:["$price",0,1]}
]
}
}
##输出结果
{_id:1,"costByDollar":true}
{_id:1,"costByDollar":false}
3、Boolean表达式
表达式返回true或者false，目前支持三种操作$and、$or、$not；语义基本与query中相同操作符一致。具体参见【Boolean】
4、比较表达式（Comparison）
语义基本同query，只是语法不同，它们需要返回true、false，比如：$eq、$gt等等。参见【Comparison】
5、集合操作（Set）
对于多个Arrays进行比较操作:
1）$setEquals：如果输入的两个或者多个Arrays中都持有相同的Set（不同值的列表）则返回true，否则返回false。
2）$setIntersection：返回输入的多个Arrays的交集。
3）$setUnion：返回并集。
5）$setDifference：返回差集，即在第一个Set中存在，但在第二个Set中不存在的数组元素。只接收2个数组参数。
6）$setIsSubset：如果第一个Set为第二个Set的子集（包括完全相同）则返回true，否则返回false。
7）$anyElementsTrue：如果此Set中有任何一个值为true，则返回true，否则返回false。只接收一个Array参数。
8）$allElementsTrue：只有当所有的元素都为true时，才返回true。只接收一个Array参数。
6、数学计算
对数字类型的字段值进行计算，比如$add（+），$subtract（-），$multiply（*），$divide（/），$mod（%）,3.2版本新增计算表达式；语法都一样，例如：
$project : {
totalAmount : {$multiply: {"$price","$qunatity"}}
}
7、字符串操作
对字符串进行操作，非常简便有用的方法：
1）$concat：拼接字符串，此操作可以接收多个参数列表。
$concat:["$lastname",",","$firstname"]
2）$substr：截取字符串，语法：$substr : [<input expression>,<start>,<length>]，其中start默认起始为0。
3）$toLower、$toUper：大小写转换
4）$strcasecmp：不分大小写，进行字符串比较，如果2个字符串一样，则返回0；如果第一个字符串比第二个“大”（字典顺序），则返回1，否则返回false。
8、$size：计算数组的元素个数，语法：$size : <array expression>；3.2版本新增数组表达式
9、日期计算
对Date或者timestamp进行计算，比如获取时间的月份、时间戳等等。简单列举如下：
$dayOfYear：当前日期是此年的第几天，0~366（闰年）；$dayOfMonth：当前日期是本月的第几天，0~31；$dayOfWeek：当前日期是本周的第几天，1~7，1表示周日，7表示周六；$year：返回当前日期的年份，比如2014。$month：返回当前日期的月份。$week：返回当前日期为当年的第几周，0~53周。$hour、$minute、$second、$millisecond表示获取当前日期的时、分、秒、毫秒部分。
$dateToString：日期的格式转换，这个比较常用，语法格式：$dataToString : {format:<format>,date:"$date"}。格式占位符如下：
%Y：年份，4个数字，比如2014；%m：月份，2个数字，01~12；%d：日期，2个数字，01~31；%H：小时，2个数字，00~23；%M：分钟，00~59；%S：秒，00~59；%L：毫秒，000~999；%j：当前day为年的第几天，001~366；%U：当前日期所在年的第几周，00~53；%w：day是当前周的第几天，1~7。
$project : {
"dateAsString":{$dateToString:{format:"%Y-%m-%d %H:%M:%S",date:"$date"}}
}
10、条件表达式
1）$cond
上文中我们已经看到相关示例，它的作用等同于一个“三元表达式”，语法：$cond:{if : <boolean-expression>, then : <true-expression>,else : <false-expression> }。
2）$ifNull
判定指定表达式是否为null，比如字段不存在（undefined）、值为null。语法：$ifNull: [<expression>,<return-value expression>]，如果expression的值为null则返回“return-value”。
四、示例
通过上文，我们基本了解了aggretation的原理和使用技巧，对于复杂的数据计算通常需要多次计算才能产出结果。此外还需要强调一点，良好的数据结构和数据存储模型，对数据分析和统计具有非常大的帮助，所以我们希望大家在数据分析之前，和appliation约定数据的模型，如果数据模型相差甚远，那么需要进行一次（或者增量的）清洗数据，并转存到特定的数据仓库中（比如mongodb，hive等）。
因为aggregate和mapreduce每次任务只能操作一个collection，数据数据来自多个collection，那么就需要多次分析，将分析结果统一写入一个result collection中，然后在对result进行最终的计算。
数据校对是个非常棘手的问题，通常可以采用“取样法”来测试，比如限定一定范围（或者随机）对整体数据取样1/10，然后对这个小数据集先运行数据统计脚本，查看结果是否有异动，对于异常情况适度调整算法。
接下来我们使用aggregate操作完成如下数据统计指标：
1）横向维度：统计当前月份之前的6个月新用户在当月的留存；比如现在为9月份，统计出3~8月份注册的用户，在9月份的留存情况。
2）纵向维度：分平台，分别统计“android”、“ios”、“pc”三个平台在横向维度的数据。
期望数据结果：
------------------------------------------------------------- | 平台\月份 | 1 | 2 | 3 | 4 | 5 | 6 | ------------------------------------------------------------- | android | 100 | 80 | 100 | 150 | 160 | 200 | ------------------------------------------------------------- | ios | 100 | 80 | 100 | 150 | 160 | 200 | ------------------------------------------------------------- | pc | 100 | 80 | 100 | 150 | 160 | 200 | -------------------------------------------------------------
我们假定有个collection为user，当用户注册时新建一条记录，created表示用户注册的时间，modfied表示用户最后登录的时间；userid上创建唯一索引，created、modified创建了组合索引，原始数据格式：
{_id : 1,userid:"10010","platform":"android","created":{"$date":1448805204482},"modified":{"$date":1448805204482}
aggregate语句：
db.user.aggregate([
{ "$match" : {
"created" : {
"$gte" : { "$date" : 1430379347719 }, "$lt" : { "$date" : 1446307200000 } },
"modified" : { "$gte" : { "$date" : 1446190547719 } }
}
},
{ "$group" : {
"_id" : { "month" : { "$month" : "$created" },
"platform" : "$platform" },
"total" : { "$sum" : 1 }
}
},
{ "$project" : {
"platform" : "$_id.platform",
"total" : 1,
"month" : "$_id.month",
"_id" : 0
}
},
{ "$sort" : { "platform" : 1, "month" : 1 } },
{ "$group" : {
"_id" : "$platform",
"result" : { "$push" : { "month" : "$month", "total" : "$total" } }
}
},
{ "$match" : {
"created" : { "$gte" : { "$date" : 1430379347719 }, "$lt" : { "$date" : 1446307200000 } },
"modified" : { "$gte" : { "$date" : 1446190547719 } }
}
},
{ "$group" : {
"_id" : { "month" : { "$month" : "$created" }, "platform" : "$platform" },
"total" : { "$sum" : 1 }
}
},
{ "$project" : {
"platform" : "$_id.platform",
"total" : 1,
"month" : "$_id.month",
"_id" : 0
}
},
{ "$sort" : { "platform" : 1, "month" : 1 } }
]
}
JAVA程序实例：
MongoCollection<Document> collection = db.getCollection("user");
Date now = new Date();//11月30日
List<Document> pipeline = new ArrayList<Document>();
Document filter = new Document();
//过滤：
//1)注册时间在6各月之前，不包括当月；
//2)最后登陆时间在当月
filter.put("created",
new Document("$gte",DateUtils.addMonths(now, -7)).append("$lt", DateUtils.truncate(now, Calendar.MONTH)));
filter.put("modified",new Document("$gte",DateUtils.addMonths(now, -1)));
Document match = new Document("$match",filter);
pipeline.add(match);
Document group = new Document();
//第一次group，将“月份”和“平台”作为复合key
group.put("$group", new Document("_id", new Document("month", new Document("$month", "$created")).append("platform", "$platform"))
.append("total", new Document("$sum", 1)));
pipeline.add(group);
//重命名字段名
Document project = new Document("$project",new Document("platform","$_id.platform")
.append("total",1).append("month","$_id.month").append("_id",0));
pipeline.add(project);
//按照平台和月份排序，确保，同一个平台的数据，是按照月份排序的，因为接下来要使用$push
Document sort = new Document("$sort",new Document("platform",1).append("month", 1));
pipeline.add(sort);
//第二次group，将上述输出数据，转换成根据平台为维度的inline方式。便于页面展示。
Document finalGroup = new Document("$group",new Document("_id","$platform")
.append("result", new Document("$push", new Document("month", "$month").append("total", "$total"))));
pipeline.add(finalGroup);
//pipeline.add(new Document("$out","_" + DateFormatUtils.format(now,"yyyyMMdd")));
MongoCursor<Document> cursor = collection.aggregate(pipeline).allowDiskUse(true).iterator();
while (cursor.hasNext()) {
Document item = cursor.next();
System.out.println(item.toJson());
}
cursor.close();
输出结果示例：
{
"_id": "pc",
"result": [
{
"month": 5,
"total": 2736
},
{
"month": 6,
"total": 2877
},
{
"month": 7,
"total": 2718
},
{
"month": 8,
"total": 2791
},
{
"month": 9,
"total": 2887
},
{
"month": 10,
"total": 2800
}
]
}
{
"_id": "android",
"result": [
{
"month": 5,
"total": 2742
},
{
"month": 6,
"total": 2850
},
{
"month": 7,
"total": 2782
},
{
"month": 8,
"total": 2752
},
{
"month": 9,
"total": 2740
},
{
"month": 10,
"total": 2941
}
]
}
非常有趣的是，aggreate可以支持2个group、甚至更多个group操作，这大大减少了数据分析的开发难度，如果使用mapreduce可能需要开发多个脚本且运行多次。在技术日新月异的今天，基于大数据平台的数据计算技术不断涌现，出现大量了“SQL-LIKE Engineer”，它们可以将SQL语句转换成适用于NOSQL平台的操作，比如将SQL语句转换成Mongodb的aggreation或者mapreduce（内部或将数据转存到hadoop上进行mapreduce），这样又进一步的提升了开发效率，我们稍后介绍Apache Drill：mongodb平台上的SQL-Like引擎。
相关推荐
Spring Data MongoDB中文文档 便宜下载了。
mongodb aggregation运用文档解释各个方法的使用例子
spring_mongodb_aggregation_grouping 这是一个用于在mongodb中显示聚合分组的简单项目
MongoDB新技术特性与展望.pptx
MongoDB 聚合管道（Aggregation Pipeline） - 张善友 - 博客园
主要介绍了MongoDB中强大的统计框架Aggregation使用实例解析,文中举了Python和Java使用Aggregation的数据统计例子进行讲解,需要的朋友可以参考下
本课程是一套关于MongoDB应用开发的实战性教程，名为《深入浅出MongoDB应用实战开发(基础、开发指南、系统管理、集群及系统架构）》，教程侧重于讲解MongoDB的常用特性及高级特性，从实际开发的角度出发对MongoDB...
Spring Data MongoDB 2.1中的新特性 5.2。Spring Data MongoDB 2.0中的新特性 5.3。Spring Data MongoDB 1.10中的新特性 5.4。Spring Data MongoDB 1.9中的新特性 5.5。Spring Data MongoDB 1.8中的新特性 5.6。...
MongoDB操作手册详细介绍MongoDB的语法用法
mongoDB中文文档和一个可以运行的dome,其中包括mongoDB索引，管道，事务，集合和crud等基本操作
match：过滤文档使用$ project 使用表达式光标式方法阶段$ group阶段$放开$ lookup阶段利用MongoDB的灵活架构mongoimport 模式的重要性探索模式迁移架构观看次数用累加器补充架构单个文档中的树状数据富有表现力的...
MongoDB 3.6 中文文档
使用java向mongodb中插入数据 一、Linux安装mongodb 二、MongoDB客户端工具(MongoVUE)访问mongodb 三、通过Java操作MongoDB
Spring Data MongoDB 中文参考文档和Springboot使用例子，其中还介绍了spring-data-mongodb 如何使用mongodb的全文检索。
MongoDB: The Definitive Guide by Kristina Chodorow and Michael Dirolf Copyright © 2010 Kristina Chodorow and Michael Dirolf. All rights reserved.
mongodb中文API及分布式分片实例详解。
NoSQLBooster 管理MongoDB的工具。这个工具还有个曾用名--mongobooster
Spring Data MongoDB 2.1中的新特性 5.2。Spring Data MongoDB 2.0中的新特性 5.3。Spring Data MongoDB 1.10中的新特性 5.4。Spring Data MongoDB 1.9中的新特性 5.5。Spring Data MongoDB 1.8中的新特性 5.6。...