MongoDB 中主要有三个函数：
-
group：分组统计 -
aggregate：简单聚合 -
mapReduce：强大统计
1 group
1.1 定义
MongoDB 中使用group来进行分组聚合，语法如下：
db.collection.group(document);
document中信息：
{
key:{key1:true,key2:true},
cond:{},
reduce:function(curr,result){},
initial:{},
finalize:function(curr,result){}
}
document中字段说明：
-
key：分组字段，作为分组的key,等价于sql中group by a,b -
cond：查询条件 -
reduce：聚合函数
一个聚合函数操作文档的分组操作期间。这些函数可以返回一个sum或count。
该函数接受两个参数：当前文档和这个群体聚集的结果文档 -
initial：初始化聚合结果文档变量，为空时自动为每列提供初始变量 -
finalize：统计一组后的回调函数
在db.collection.group()返回最终结果之前，此功能可以修改的结果文档或替换的结果文档作为一个整体
注意：
-
group需要我们手写聚合函数的业务逻辑 -
group不支持shard cluster，无法分布式运算 - 有些版本不支持
group函数，可以通过db.collection.help()函数查看
1.2 操作group
1.2.1 求数目
db.collection.group({
key:{cat_id:1},
cond:{},
reduce:function(curr,result) {
result.cnt += 1;
},
initial:{cnt:0}
})
类似于sql:
select count(*) from goods group by cat_id;
1.2.2 求总和
db.collection.group({
key:{cat_id:1},
cond:{},
reduce:function(curr,result) {
result.num += curr.goods_number;
},
initial:{num:0}
})
类似于sql:
select sum(goods_number) from goods group by cat_id;
1.2.3 求最大
db.collection.group({
key:{cat_id:1},
cond:{},
reduce:function(curr , result) {
if(curr.shop_price > result.max) {
result.max = curr.shop_price;
}
},
initial:{max:0}
})
类似于sql:
select max(shop_price) from goods group by cat_id;
1.2.4 求平均
求平均时主要用到了finalize函数
db.collection.group({
key:{cat_id:1},
cond:{},
reduce:function(curr , result) {
result.cnt += 1; //统计数量
result.sum += curr.shop_price; //统计总量
},
initial:{sum:0,cnt:0},
finalize:function(result) {
result.avg = result.sum/result.cnt;//求平均
}
})
2 aggregate
2.1 定义
MongoDB 中聚合(aggregate)主要用于处理数据(诸如统计平均值，求和等)，并返回计算后的数据结果
基本语法：
db.collection.aggregate(docment);
具体方法实例有：$match，$project，$group，$unwind，$sort，$limit，$skip等相关操作
2.2 操作aggregate
准备脚本
> db.book.insertMany([
{"_id": 1, "author": "monday", "book": "《Java》", "like": 10},
{"_id": 2, "author": "monday", "book": "《Java Core》", "like": 20},
{"_id": 3, "author": "mengday", "book": "《Spring Boot》", "like": 15}
])
> db.book.find()
{ "_id" : 1, "author" : "monday", "book" : "《Java》", "like" : 10 }
{ "_id" : 2, "author" : "monday", "book" : "《Java Core》", "like" : 20 }
{ "_id" : 3, "author" : "mengday", "book" : "《Spring Boot》", "like" : 15 }
2.2.1 $match筛选
用于过滤数据，只输出符合条件的文档。$match使用MongoDB的标准查询操作。
筛选条件，相当于SQL中的where部分，过滤掉不满足条件的文档,可以使用常规的查询操作符，如 $gt、$lt、$in等。
> db.book.aggregate({"$match": {"like": {"$gt": 10}}})
{ "_id" : 2, "author" : "monday", "book" : "《Java Core》", "like" : 20 }
{ "_id" : 3, "author" : "mengday", "book" : "《Spring Boot》", "like" : 15 }
当match在group之前类似于SQL的where操作，在group之后类似于SQL的having操作
db.book.aggregate([{"$match": {"like": {"$gte": 10}}},{$group:{_id:'$author',total:{$sum:1}}},{$match:{total:{$gte:2}}}])
{ "_id": "monday", "total": 2}
2.2.2 $project映射
$project：修改输入文档的结构。可以用来重命名、增加或删除域，也可以用于创建计算结果以及嵌套文档。
映射相当于SQL中的 select 映射 from之间的部分，用于指定要查询的字段，或者对字段进行一些处理。
投射常用的3个功能：
- 对字段重命名。
- 在投射中使用一些表达式对字段值进行处理:数学表达式、日期表达式、字符串表达式、逻辑表达式(比较表达式、布尔表达式、控制语句)。
- 用于包含、排除字段: 设置要查询或者要过滤掉的字段，
- 0: 要过滤掉的字段，不显示
- 1：需要查询的字段。
2.2.2.1 自定义字段
查询自定义的字段
db.book.aggregate({"$project": {"_id": 0, "book": 1, "like": 1}})
{ "book" : "《Java》", "like" : 10 }
{ "book" : "《Java Core》", "like" : 20 }
{ "book" : "《Spring Boot》", "like" : 15 }
2.2.2.2 起别名
对字段起别名，相当于SQL中的as的作用
db.book.aggregate({"$project": {"_id": 0, "book name": "$book", "like": 1}})
{ "like" : 10, "book name" : "《Java》" }
{ "like" : 20, "book name" : "《Java Core》" }
{ "like" : 15, "book name" : "《Spring Boot》" }
2.2.2.3 引用原字段：$字段名
如果对_id进行起别名会将别名作为一个新的字段加入到文档中，而_id保持不变
db.book.aggregate({"$project": {"id": "$_id", "book name": "$book", "like": 1}})
{ "_id" : 1, "like" : 10, "id" : 1, "book name" : "《Java》" }
{ "_id" : 2, "like" : 20, "id" : 2, "book name" : "《Java Core》" }
{ "_id" : 3, "like" : 15, "id" : 3, "book name" : "《Spring Boot》" }
2.2.2.4 使用算术表达式
使用算术表达式 $add、$subtract、$multiply、$divide、$mod 处理数字类型的字段
db.book.aggregate({"$project": {"like": {"$add": ["$like", "$like", 1]}}})
{ "_id" : 1, "like" : 21 }
{ "_id" : 2, "like" : 41 }
{ "_id" : 3, "like" : 31 }
2.2.2.5 字符串截取
$substrCP: [exp, startOffset, numToReturn] : 字符串截取操作
db.book.aggregate({"$project": {"newValue": {"$substrCP": ["$book", 1, 4]}}})
{ "_id" : 1, "newValue" : "Java" }
{ "_id" : 2, "newValue" : "Java" }
{ "_id" : 3, "newValue" : "Spri" }
2.2.2.6 字符串拼接
$concat：[exp1, exp2, ..., expN]: 字符串操作：将数组中的多个元素拼接在一起
db.book.aggregate({"$project": {"newValue": {"$concat": ["$book", "(", "$author", ")"]}}})
{ "_id" : 1, "newValue" : "《Java》(monday)" }
{ "_id" : 2, "newValue" : "《Java Core》(monday)" }
{ "_id" : 3, "newValue" : "《Spring Boot》(mengday)" }
2.2.2.7 大小写转换
$toUpper: 字符串操作，转大写
$toLower: exp, 字符串转小写
db.book.aggregate({"$project": {"newValue": {"$toUpper": "$book"}}})
{ "_id" : 1, "newValue" : "《JAVA》" }
{ "_id" : 2, "newValue" : "《JAVA CORE》" }
{ "_id" : 3, "newValue" : "《SPRING BOOT》" }
2.2.2.8 日期表达式
日期表达式：用于获取日期中的任意一部分，年月日时分秒 星期等
$year、$month、$dayOfMonth、$dayOfWeek、$dayOfYear、$hour、$minute、$second
-
$dayOfYear: 返回该日期是这一年的第几天（全年 366 天）。 -
$dayOfMonth: 返回该日期是这一个月的第几天（1到31）。 -
$dayOfWeek: 返回的是这个周的星期几（1：星期日，7：星期六）。 -
$year: 返回该日期的年份部分。 -
$month： 返回该日期的月份部分（ 1 到 12）。 -
$week： 返回该日期是所在年的第几个星期（ 0 到 53）。 -
$hour： 返回该日期的小时部分。 -
$minute: 返回该日期的分钟部分。 -
$second: 返回该日期的秒部分（以0到59之间的数字形式返回日期的第二部分，但可以是60来计算闰秒）。 -
$millisecond：返回该日期的毫秒部分（ 0 到 999）。 -
$dateToString： { $dateToString: { format: , date: } }。
准备脚本
db.book.find()
{ "_id" : 1, "author" : "monday", "book" : "《Java》", "like" : 10, "publishDate" : ISODate("2021-04-12T13:38:14.829Z") }
{ "_id" : 2, "author" : "monday", "book" : "《Java Core》", "like" : 20, "publishDate" : ISODate("2021-04-12T13:38:14.829Z") }
{ "_id" : 3, "author" : "mengday", "book" : "《Spring Boot》", "like" : 15, "publishDate" : ISODate("2021-04-12T13:38:14.829Z") }
求日期
db.book.aggregate({"$project": {"year": {"$year": "$publishDate"}}})
{ "_id" : 1, "year" : 2021 }
{ "_id" : 2, "year" : 2021 }
{ "_id" : 3, "year" : 2021 }
2.2.2.9 比较非字符串大小
$cmp: [exp1, exp2]: 用于比较两个非字符串类型的值，exp1 == exp2 返回 0， 小于返回一个负数，大于返回一个正数
db.book.aggregate({"$project": {"result": {"$cmp": ["$like", 15]}}})
{ "_id" : 1, "result" : -1 }
{ "_id" : 2, "result" : 1 }
{ "_id" : 3, "result" : 0 }
2.2.2.10 比较字符串大小
$strcasecmp: 用于比较字符串, 不区分大小写，相等返回0
db.book.aggregate({"$project": {"result": {"$strcasecmp": ["$author", "Monday"]}}})
{ "_id" : 1, "result" : 0 }
{ "_id" : 2, "result" : 0 }
{ "_id" : 3, "result" : -1 }
2.2.2.11 比较表达式
$eq: 用于判断两个表达式是否相等，相等返回true，不相等返回false, 区分大小写
$ne: 不相等
$gt: 大于
$gte： 大于等于
$lt： 小于
$lte： 小于等于
注意：{"$eq": ["$author", "monday"]}中$eq键的值是个数组而不是一个对象
db.book.aggregate({"$project": {"result": {"$eq": ["$author", "monday"]}}})
{ "_id" : 1, "result" : true }
{ "_id" : 2, "result" : true }
{ "_id" : 3, "result" : false }
2.2.2.12 And连接
$and:[exp1, exp2, ..., expN]用于连接多个条件，当所有条件为真的时候为true
db.book.aggregate({"$project": {"result": {"$and": [{"$eq": ["$author", "monday"]}, {"$gt": ["$_id", 1]}]}}})
{ "_id" : 1, "result" : false }
{ "_id" : 2, "result" : true }
{ "_id" : 3, "result" : false }
2.2.2.13 Or连接
$or: [exp1, exp2, ..., expN]有一个为真则为真
db.book.aggregate({"$project": {"result": {"$or": [{"$eq": ["$author", "monday"]}, {"$eq": ["$_id", 1]}]}}})
{ "_id" : 1, "result" : true }
{ "_id" : 2, "result" : true }
{ "_id" : 3, "result" : false }
2.2.2.14 not连接
$not: exp 用于取反操作
db.book.aggregate({"$project": {"result": {"$not": {"$eq": ["$author", "monday"]}}}})
{ "_id" : 1, "result" : false }
{ "_id" : 2, "result" : false }
{ "_id" : 3, "result" : true }
2.2.2.15 三位运算符
$cond: [booleanExp, trueExp, falseExp]: 三位运算符
db.book.aggregate({"$project": {"result": {"$cond": [ {"$eq": ["$author", "monday"]}, "M", "F" ]}}})
{ "_id" : 1, "result" : "M" }
{ "_id" : 2, "result" : "M" }
{ "_id" : 3, "result" : "F" }
2.2.2.16 ifNull连接
$ifNull: [expr, replacementExpr]: 如果字段不存在或者字段值为null会返回replacementExpr，否则返回原来的值，即字段为空时给一个默认值
IFNULL(bool, default_value)
db.book.aggregate({"$project": {"result": {"$ifNull": ["$price", "0.00"]}}})
{ "_id" : 1, "result" : 66 }
{ "_id" : 2, "result" : "0.00" }
{ "_id" : 3, "result" : "0.00" }
2.2.3 $group 分组
2.2.3.1 单字段分组
相当于SQL中的group by 部分
$group 表示分组
_id: 用于指定要分组的字段
count: 是聚合后的结果的字段别名，类似于SQL中的as后面的别名，可以任意定义，字段前面使用$表示应用某个字段而不是一个普通的字符串
$sum: 对分组中的每个文档做什么类型的聚合，是求和还是求平均数等，1：表示对分组的每一条文档都加1进行统计，这就相当于SQL中的count(*)
select _id, count(*) from book group by author
db.book.aggregate({"$group": {"_id": "$author", "count": {"$sum": 1}}})
{ "_id" : "monday", "count" : 2 }
{ "_id" : "mengday", "count" : 1 }
2.2.3.2 对多个字段进行分组
类似于：select _id, count(*) from book group by author, like
db.book.aggregate({"$group": {"_id": {"author": "$author", "like": "$like"}, "count": {"$sum": 1}}})
{ "_id" : { "author" : "monday", "like" : 10 }, "count" : 1 }
{ "_id" : { "author" : "monday", "like" : 20 }, "count" : 1 }
{ "_id" : { "author" : "mengday", "like" : 15 }, "count" : 1 }
2.2.3.3 聚合运算
$avg: 求分组中某个字段的平均值
$max: 求分组中某个字段的最大值
$min: 求分组中某个字段最小的值
$first: 求分组中的第一个值
$last: 求分组中最后一个值
db.book.aggregate( {"$group": {"_id": "$author", "avg": {"$avg": "$like"}}} )
{ "_id" : "monday", "avg" : 15 }
{ "_id" : "mengday", "avg" : 15 }
2.2.3.4 addToSet
$addToSet: 字段引用, 将分组后的每个文档指定的值放在set集合中，集合不重复，无序
db.book.aggregate( {"$group": {"_id": "$author", "likes": {"$addToSet": "$like"}}} )
{ "_id" : "monday", "likes" : [ 20, 10 ] }
{ "_id" : "mengday", "likes" : [ 15 ] }
2.2.3.5 push操作
$push: exp, 将分组后的每个文档指定的值放在数组中，允许重复，有序
db.book.aggregate( {"$group": {"_id": "$author", "likes": {"$push": "$like"}}} )
{ "_id" : "mengday", "likes" : [ 15 ] }
{ "_id" : "monday", "likes" : [ 10, 20 ] }
2.2.4 $unwind
将数组的每一个元素都单独作为一条文档进行拆分。
db.comments.insert({"_id": 1, "title": "java", "comment": ["good", "very good"]})
db.comments.find()
{ "_id" : 1, "title" : "java", "comment" : [ "good", "very good" ] }
db.comments.aggregate( {"$unwind": "$comment"} )
{ "_id" : 1, "title" : "java", "comment" : "good" }
{ "_id" : 1, "title" : "java", "comment" : "very good" }
2.2.5 $sort
对文档进行排序，相当于SQL中的order by。
db.book.aggregate( {"$sort": {"_id": -1}} )
{ "_id" : 3, "author" : "mengday", "book" : "《Spring Boot》", "like" : 15, "publishDate" : ISODate("2021-04-12T13:38:14.829Z") }
{ "_id" : 2, "author" : "monday", "book" : "《Java Core》", "like" : 20, "publishDate" : ISODate("2021-04-12T13:38:14.829Z"), "price" : null }
{ "_id" : 1, "author" : "monday", "book" : "《Java》", "like" : 10, "publishDate" : ISODate("2021-04-12T13:38:14.829Z"), "price" : 66 }
2.2.6 $limit
限制返回的条数，相当于SQL中的limit count 语句。
db.book.aggregate( {"$limit": 2} )
{ "_id" : 1, "author" : "monday", "book" : "《Java》", "like" : 10, "publishDate" : ISODate("2021-04-12T13:38:14.829Z"), "price" : 66 }
{ "_id" : 2, "author" : "monday", "book" : "《Java Core》", "like" : 20, "publishDate" : ISODate("2021-04-12T13:38:14.829Z"), "price" : null }
2.2.7 $skip
跳过前N条文档，和limit结合可用于分页。
db.book.aggregate( {"$skip": 2}, {"$limit": 2} )
{ "_id" : 3, "author" : "mengday", "book" : "《Spring Boot》", "like" : 15, "publishDate" : ISODate("2021-04-12T13:38:14.829Z") }
2.2.8 综合使用
聚合框架，就是将上一个操作符处理的结果交个下一个操作符继续处理（这就是Linux中的管道操作），可以使用任意多个操作符，同一个操作符也可以使用多次
- 首先通过
$match过滤掉不匹配的文档， - 接着讲满足条件的文档交给
$group进行分组， - 分组后将将分组后的结果交给
$sort进行排序， - 然后将排序后的结果交给
$skip处理，跳过前几条， - 把剩下的文档交给
$limit处理，获取最终的聚合结果。
db.book.aggregate(
{"$match": {"like": {"$gte" : 10} }},
{"$group": {"_id": "$author", "count": {"$sum": 1}}},
{"$sort": {"count": -1}},
{"$skip": 1},
{"$limit": 1}
)
{ "_id" : "mengday", "count" : 1 }
2.3 常见聚合表达式
下表展示了一些聚合的计算表达式:
| 表达式 | 描述 | 实例 |
|---|---|---|
$sum |
计算总和 | db.mycol.aggregate([{$group : {_id : "$by_user", num_tutorial : {$sum : "$likes"}}}]) |
$avg |
计算平均值 | db.mycol.aggregate([{$group : {_id : "$by_user", num_tutorial : {$avg : "$likes"}}}]) |
$min |
获取集合中所有文档对应值得最小值 | db.mycol.aggregate([{$group : {_id : "$by_user", num_tutorial : {$min : "$likes"}}}]) |
$max |
获取集合中所有文档对应值得最大值。 | db.mycol.aggregate([{$group : {_id : "$by_user", num_tutorial : {$max : "$likes"}}}]) |
$push |
将值加入一个数组中，不会判断是否有重复的值 | db.mycol.aggregate([{$group : {_id : "$by_user", url : {$push: "$url"}}}]) |
$addToSet |
将值加入一个数组中，会判断是否有重复的值，若相同的值在数组中已经存在了，则不加入。 | db.mycol.aggregate([{$group : {_id : "$by_user", url : {$addToSet : "$url"}}}]) |
$first |
根据资源文档的排序获取第一个文档数据。 | db.mycol.aggregate([{$group : {_id : "$by_user", first_url : {$first : "$url"}}}]) |
$last |
根据资源文档的排序获取最后一个文档数据 | db.mycol.aggregate([{$group : {_id : "$by_user", last_url : {$last : "$url"}}}]) |
3 mapReduce
Map-Reduce是一种计算模型，简单的说就是将大批量的工作（数据）分解（MAP）执行，然后再将结果合并成最终结果（REDUCE）。
MongoDB提供的Map-Reduce非常灵活，对于大规模数据分析也相当实用。
3.1 mapReduce语法
以下是MapReduce的基本语法：
db.collection.mapReduce(
function() {emit(key,value);}, //map 函数
function(key,values) {return reduceFunction}, //reduce 函数
{
out: collection,
query: document,
sort: document,
limit: number
}
)
使用 MapReduce 要实现两个函数Map 函数和 Reduce 函数,Map 函数调用 emit(key, value), 遍历 collection 中所有的记录, 将 key 与 value 传递给 Reduce 函数进行处理。
Map 函数必须调用 emit(key, value)返回键值对。
参数说明:
-
map：映射函数 (生成键值对序列,作为reduce函数参数)
把属于同一个组的数据，映射到一个数组上 -
reduce：统计函数，把数组(同一组)的数据，进行运算
reduce函数的任务就是将key-values变成key-value，也就是把values数组变成一个单一的值value -
out：统计结果存放集合 (不指定则使用临时集合,在客户端断开后自动删除)。 -
query：一个筛选条件，只有满足条件的文档才会调用map函数。（query。limit，sort可以随意组合） -
sort和limit结合的sort排序参数（也是在发往map函数前给文档排序），可以优化分组机制 -
limit发往map函数的文档数量的上限（要是没有limit，单独使用sort的用处不大）
以下实例在集合 orders 中查找 status:"A" 的数据，并根据 cust_id 来分组，并计算 amount 的总和。
image.png
3.2 与分组聚合对比
与Aggregate中引用集合中字段区别：
- 在分组聚合
Aggregate函数中，引用原字段用$原字段 - 在
MapReduce中引用原来字段用this，比如：this.age
3.3 使用 mapReduce
准备脚本
db.user.insertMany([
{"name" : "鲁迅","book" : "呐喊","price" : 38.0,"publisher" : "人民文学出版社"},
{"name" : "曹雪芹","book" : "红楼梦","price" : 22.0,"publisher" : "人民文学出版社"},
{"name" : "钱钟书","book" : "宋诗选注","price" : 99.0,"publisher" : "人民文学出版社"},
{"name" : "钱钟书","book" : "谈艺录","price" : 66.0,"publisher" : "三联书店"},
{"name" : "鲁迅","book" : "彷徨","price" : 55.0,"publisher" : "花城出版社"}
]);
查询每个作者的总计价格
db.user.mapReduce(
function(){emit(this.name,this.price)},
function(key,value){return Array.sum(value)},
{out:"totalPrice"});
查看处理结果
db.totalPrice.find();
{
"_id": "鲁迅",
"value": 22
}
// 2
{
"_id": "钱钟书",
"value": 93
}
// 3
{
"_id": "曹雪芹",
"value": 165
}
或者使用引用的方法（查询每个价格在 40以上的书，并用逗号分隔）
所有函数写在一起
db.user.mapReduce(
function(){emit(this.name,this.book)},
function(key,value){return value.join(',')},
{query:{price:{$gt:40}},out:"books"})
db.books.find();
单独写，然后引用
var map=function(){emit(this.name,this.book)}
var reduce=function(key,value){return value.join(',')}
var options={query:{price:{$gt:40}},out:"books"}
db.user.mapReduce(map,reduce,options);
3.4 runCommand
3.4.1 语法
db.runCommand(
{
mapReduce: <collection>,
map: <function>,
reduce: <function>,
finalize: <function>,
out: <output>,
query: <document>,
sort: <document>,
limit: <number>,
scope: <document>,
jsMode: <boolean>,
verbose: <boolean>,
bypassDocumentValidation: <boolean>,
collation: <document>
}
)
参数含义：
-
mapReduce：表示要操作的集合 -
map：map函数 -
reduce：reduce函数 -
finalize：最终处理函数 -
out：输出的集合 -
query：对结果进行过滤 -
sort：对结果排序 -
limit：返回的结果数 -
scope：设置参数值，在这里设置的值在map，reduce，finalize函数中可见 -
jsMode：是否将地图执行的中间数据由javascript对象转换成BSON对象，替换为false -
verbose：是否显示详细的时间统计信息 -
bypassDocumentValidation：是否绕过文档验证 -
collation：其他一些校对
3.4.2 案例
如下操作，表示执行MapReduce操作重新统计的集合限制返回条数，限制返回条数之后再进行统计操作，如下：
var map=function(){emit(this.name,this.book)}
var reduce=function(key,value){return value.join(',')}
db.runCommand({mapreduce:'user',map,reduce,out:"books",limit:4,verbose:true})
db.books.find()
执行结果：
{ "_id" : "鲁迅", "value" : "呐喊" }
{ "_id" : "曹雪芹", "value" : "红楼梦" }
{ "_id" : "钱钟书", "value" : "谈艺录,宋诗选注" }
这里进行对比发现，因为limit的原因，鲁迅的第一本书不见了
finalize操作表示最终处理函数，如下：f1函数的第一个参数键表示emit中的第一个参数，第二个参数表示reduce的执行结果，我们可以在f1中对这个结果进行再处理
var f1 = function(key,reduceValue){var obj={};obj.author=key;obj.books=reduceValue; return obj}
var map=function(){emit(this.name,this.book)}
var reduce=function(key,value){return value.join(',')}
db.runCommand({mapreduce:'user',map,reduce,out:"books",finalize:f1})
db.books.find()
执行结果：
{ "_id" : "鲁迅", "value" : { "author" : "鲁迅", "books" : "彷徨,呐喊" } }
{ "_id" : "曹雪芹", "value" : { "author" : "曹雪芹", "books" : "红楼梦" } }
{ "_id" : "钱钟书", "value" : { "author" : "钱钟书", "books" : "谈艺录,宋诗选注" } }
