☰
×
MongoDB 中 MapReduce 使用
  • 2024-02-14 15:22:22
  • 2422 热度
  • 0 评论

玩过 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用法详解(未找到原始出处)

alay

Flame

Hello world!

0 评论
留下评论
作者
Flame
Flame
Hello world!
私信 Ta
热门分类
后端开发 863 前端开发 58 其它 44
热门标签