MongoDB
10 分钟 读完 (大约 1514 个字)

MongoDB 中 MapReduce 使用

玩过 Hadoop 的小伙伴对 MapReduce 应该不陌生，MapReduce 的强大且灵活，它可以将一个大问题拆分为多个小问题，将各个小问题发送到不同的机器上去处理，所有的机器都完成计算后，再将计算结果合并为一个完整的解决方案，这就是所谓的分布式计算。本文我们就来看看 MongoDB 中 MapReduce 的使用。

mapReduce

MongoDB 中的 MapReduce 可以用来实现更复杂的聚合命令，使用 MapReduce 主要实现两个函数：map 函数和 reduce 函数，map 函数用来生成键值对序列， map 函数的结果作为 reduce 函数的参数，reduce 函数中再做进一步的统计，比如我的数据集如下：

1
2
3
4
5
{"_id" : ObjectId("59fa71d71fd59c3b2cd908d7"),"name" : "鲁迅","book" : "呐喊","price" : 38.0,"publisher" : "人民文学出版社"}
{"_id" : ObjectId("59fa71d71fd59c3b2cd908d8"),"name" : "曹雪芹","book" : "红楼梦","price" : 22.0,"publisher" : "人民文学出版社"}
{"_id" : ObjectId("59fa71d71fd59c3b2cd908d9"),"name" : "钱钟书","book" : "宋诗选注","price" : 99.0,"publisher" : "人民文学出版社"}
{"_id" : ObjectId("59fa71d71fd59c3b2cd908da"),"name" : "钱钟书","book" : "谈艺录","price" : 66.0,"publisher" : "三联书店"}
{"_id" : ObjectId("59fa71d71fd59c3b2cd908db"),"name" : "鲁迅","book" : "彷徨","price" : 55.0,"publisher" : "花城出版社"}

假如我想查询每位作者所出的书的总价，操作如下：

1
2
3
4
5
var map=function(){emit(this.name,this.price)}
var reduce=function(key,value){return Array.sum(value)}
var options={out:"totalPrice"}
db.sang_books.mapReduce(map,reduce,options);
db.totalPrice.find()

emit 函数主要用来实现分组，接收两个参数，第一个参数表示分组的字段，第二个参数表示要统计的数据，reduce 来做具体的数据处理操作，接收两个参数，对应 emit 方法的两个参数，这里使用了 Array 中的 sum 函数对 price 字段进行自加处理，options 中定义了将结果输出的集合，届时我们将在这个集合中去查询数据，默认情况下，这个集合即使在数据库重启后也会保留，并且保留集合中的数据。查询结果如下：

1
2
3
4
5
6
7
8
9
10
11
12
{
    "_id" : "曹雪芹",
    "value" : 22.0
}
{
    "_id" : "钱钟书",
    "value" : 165.0
}
{
    "_id" : "鲁迅",
    "value" : 93.0
}

再比如我想查询每位作者出了几本书，如下：

1
2
3
4
5
var map=function(){emit(this.name,1)}
var reduce=function(key,value){return Array.sum(value)}
var options={out:"bookNum"}
db.sang_books.mapReduce(map,reduce,options);
db.bookNum.find()

查询结果如下：

1
2
3
4
5
6
7
8
9
10
11
12
{
    "_id" : "曹雪芹",
    "value" : 1.0
}
{
    "_id" : "钱钟书",
    "value" : 2.0
}
{
    "_id" : "鲁迅",
    "value" : 2.0
}

将每位作者的书列出来，如下：

1
2
3
4
5
var map=function(){emit(this.name,this.book)}
var reduce=function(key,value){return value.join(',')}
var options={out:"books"}
db.sang_books.mapReduce(map,reduce,options);
db.books.find()

结果如下：

1
2
3
4
5
6
7
8
9
10
11
12
{
    "_id" : "曹雪芹",
    "value" : "红楼梦"
}
{
    "_id" : "钱钟书",
    "value" : "宋诗选注,谈艺录"
}
{
    "_id" : "鲁迅",
    "value" : "呐喊,彷徨"
}

比如查询每个人售价在 ￥40 以上的书：

1
2
3
4
5
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.sang_books.mapReduce(map,reduce,options);
db.books.find()

query 表示对查到的集合再进行筛选。

结果如下：

1
2
3
4
5
6
7
8
{
    "_id" : "钱钟书",
    "value" : "宋诗选注,谈艺录"
}
{
    "_id" : "鲁迅",
    "value" : "彷徨"
}

runCommand实现

我们也可以利用 runCommand 命令来执行 MapReduce。格式如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
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 是否将map执行的中间数据由javascript对象转换成BSON对象，默认为false
verbose 是否显示详细的时间统计信息
bypassDocumentValidation 是否绕过文档验证
collation 其他一些校对

如下操作，表示执行 MapReduce 操作并对统计的集合限制返回条数，限制返回条数之后再进行统计操作，如下：

1
2
3
4
var map=function(){emit(this.name,this.book)}
var reduce=function(key,value){return value.join(',')}
db.runCommand({mapreduce:'sang_books',map,reduce,out:"books",limit:4,verbose:true})
db.books.find()

执行结果如下：

1
2
3
4
5
6
7
8
9
10
11
12
{
    "_id" : "曹雪芹",
    "value" : "红楼梦"
}
{
    "_id" : "钱钟书",
    "value" : "宋诗选注,谈艺录"
}
{
    "_id" : "鲁迅",
    "value" : "呐喊"
}

小伙伴们看到，鲁迅有一本书不见了，就是因为 limit 是先限制集合返回条数，然后再执行统计操作。

finalize 操作表示最终处理函数，如下：

1
2
3
4
5
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:'sang_books',map,reduce,out:"books",finalize:f1})
db.books.find()

f1 第一个参数 key 表示 emit 中的第一个参数，第二个参数表示 reduce 的执行结果，我们可以在 f1 中对这个结果进行再处理，结果如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
{
    "_id" : "曹雪芹",
    "value" : {
        "author" : "曹雪芹",
        "books" : "红楼梦"
    }
}
{
    "_id" : "钱钟书",
    "value" : {
        "author" : "钱钟书",
        "books" : "宋诗选注,谈艺录"
    }
}
{
    "_id" : "鲁迅",
    "value" : {
        "author" : "鲁迅",
        "books" : "呐喊,彷徨"
    }
}

scope 则可以用来定义一个在 map、reduce 和 finalize 中都可见的变量，如下：

1
2
3
4
5
var f1 = function(key,reduceValue){var obj={};obj.author=key;obj.books=reduceValue;obj.sang=sang; return obj}
var map=function(){emit(this.name,this.book)}
var reduce=function(key,value){return value.join(',--'+sang+'--,')}
db.runCommand({mapreduce:'sang_books',map,reduce,out:"books",finalize:f1,scope:{sang:"haha"}})
db.books.find()

执行结果如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
{
    "_id" : "曹雪芹",
    "value" : {
        "author" : "曹雪芹",
        "books" : "红楼梦",
        "sang" : "haha"
    }
}
{
    "_id" : "钱钟书",
    "value" : {
        "author" : "钱钟书",
        "books" : "宋诗选注,--haha--,谈艺录",
        "sang" : "haha"
    }
}
{
    "_id" : "鲁迅",
    "value" : {
        "author" : "鲁迅",
        "books" : "呐喊,--haha--,彷徨",
        "sang" : "haha"
    }
}

好了，MongoDB 中的 MapReduce 我们就先说到这里，小伙伴们有问题欢迎留言讨论。

参考资料：

  1. 《MongoDB权威指南第2版》
  2. mongodb mapreduce小试
  3. mongoDB–mapreduce用法详解(未找到原始出处)
# MongoDB

喜欢这篇文章吗？扫码关注公众号【江南一点雨】，【江南一点雨】专注于 SPRING BOOT+微服务以及前后端分离技术，每天推送原创技术干货，关注后回复 JAVA，领取松哥为你精心准备的 JAVA 干货!

觉得不错，可以给小编加个鸡腿

MongoDB 副本集搭建
MongoDB 管道操作符(二)

评论

扫码关注微信公众号

分类

标签云

2019-article Actuator Blog Bug CAS CORS Chrome Docker Druid Ehcache ElasticJob ElasticSearch ElasticSerach Eureka FastDFS Flowable Flyway Freemarker Git GitHub HDC Https I18N IDEA IDEA，EasyCode IntelliJ IDEA JDK16 JWT Java JavaWeb JdbcTemplate Jedis Jenkins Jpa LiveReload Log Lombok MacBook MacBookPro Mail Maven MongoDB MyBatis MyBatisPlus MyCat MySQL MySQL8 Nacos Nginx OAuth2 PicGo Platform Plugin RabbitMQ Redis Resume SSM Shiro Spring Spring Boot Spring Cloud Spring Data Spring Security Spring6 SpringBoot SpringBoot3 SpringCloud SpringData SpringDataJpa SpringMVC SpringSecurity Swagger Swagger2 Swagger3 TLS Tencent Thymeleaf TienChin Transaction Video ViewResolver Vue Vue3 Web WebFlux blog docker domain elasticjob es gRPC hdz hexo huawei i18n idea java life mail nginx redis router saga seata tcc transaction vhr video vue vuepress weixin xa xxl-job yaml 七牛云 中间件 主从复制 事务 侃侃而谈 全局ID 分布式 分布式事务 分布式锁 前后端分离 华为 单体应用 外包 学习资源 定时任务 工具 干货汇总 延迟队列 微人事 微服务 杂谈 条件注解 死信队列 消息过期 消息驱动 源码解析 源码解读 生活 私活 程序人生 程序员 腾讯 自学 视频 视频教程 训练营 资料 逆向工程 面试

最新文章

归档

友链

Your browser is out-of-date!

Update your browser to view this website correctly. Update my browser now

×