谭明智的空间

我们一直在努力....

基于MongoDB MapReduce的统计分析

标签: MongoDB MapReduce nosql
 基于MongoDB MapReduce的统计分析
 
 
前面已经简单介绍了MongoDB在OECP社区的一个应用:动态消息的设计实现。在上次的应用中,我们只介绍了MongoDB最基本的查询的功能,今天我再介绍一下MongoDB更加高级的应用:用MongoDB做统计分析。
OECP社区中,我们为了更加准确的分析网站的访问情况,以便能够为用户更准确的推荐他们感兴趣的内容,我们需要将页面的访问记录存储下来。对于这些数据,主要由以下几个特点:
  • 与业务无关,尽量将数据存储和业务数据分离,减少业务数据库的压力。而且对数据的一致性要求不高。
  • 每当访问一个页面就要存储一条记录,实时插入操作的要求很高,当然可以使用缓存作为临时缓冲来解决数据频繁更新的问题。
  • 数据随着访问量的增长膨胀的很快,如果一个页面1天有100个PageViews,将会新增100条数据,数据量远远高于业务数据,而且要比我们上次说的消息动态的数据的数量级要大得多。网站要尽量存储至少两个月的数据,当网站访问量很大的时候,要解决的是海量数据的存储。
所以从存储上考虑,我们依然选择了MongoDB作为持久存储。由于NoSQL数据库在数据查询的多样性能力太低,特别是标准的Key-Value数据库,一般的做法就是用NoSQL负责日志的存储,分析需要将数据抽取到关系数据库中再进行统计查询。但是MongoDB却提供给我们非常丰富的查询统计功能,group 和MapReduce都能实现SQL中group by,sum,count之类的统计查询分析。Group的功能已经可以实现简单的统计功能,但是当数据量非常大的时候,group处理能力就不太好了,所以我们一开始就使用MapReduce进行统计分析。
先看一下官方对MapReduce的介绍:
db.runCommand(
 { mapreduce : <collection>,
   map : <mapfunction>,
   reduce : <reducefunction>
   [, query : <query filter object>]
   [, sort : <sort the query.  useful foroptimization>]
   [, limit : <number of objects to returnfrom collection>]
   [, out : <output-collection name>]
   [, keeptemp: <true|false>]
   [, finalize : <finalizefunction>]
   [, scope : <object where fields go into javascript global scope >]
   [, verbose : true]
 }
);
而java驱动下提供的方法主要有两个:
DBCollection.mapReduce(String map, String reduce,String outputCollection,
                  DBObject query);
DBCollection.mapReduce(DBObject command);//该接口按照上面的介绍,总是报错,不知道此该如何应用
 
PV数据存储结构:(这些属性主要是为了支持我们以后根据各种维度去分析)
entityId:实体ID,
entityName:实体名称,
userid:(登录)访问者ID,
sessionId:会话ID,
referer:来源URL,
url:当前页面url,
title:显示的标题,
date:访问时间,
ip:访问者IP
 
第一个应用场景:当访问某用户的空间时,得到某用户最新的访问记录,同一个页面重复访问的话,返回最新的一次访问。
  • 首先是map方法,主要是定义outputCollection的结构。OutputCollection的输出结构为:{_id:key,value:value}
java 代码
  1. String mapfun = "function(){emit({url:this.url,title:this.title},this.date)}";//key={url:this.url,title:this.title},value=reduce方法的返回值。   
  • 其次是reduce方法
java 代码
  1. String reducefun = "function(key,vals){var date=0; for(var i in vals){ if(date==0){date=vals[i];}else if(vals[i]>date){date=vals[i];}} return date;}";//如果同一个key的数据,相互比较时间,将最近时间返回。   
  • 执行
java 代码
  1. DBObject query = newBasicDBObject();   
  2. query.put("userid", userid);   
  3. query.put("date", newBasicDBObject("$gte", fromDate));   
  4. *.getCollection().mapReduce(mapfun, reducefun,"pageview_results", query);//最好定义query,以降低统计的原始结果集   
  • 遍历pageview_results集合的结果:[{_id:{url:”/blog/yongtree/258”,title:’博客1’},value:’2010-10-11 20:30:56’},{_id:{url:”/blog/slx/288”,title:’博客2’}, value:’2010-10-01 02:23:33’}]
 
注意:mapfun和reducefun字符串里面是写的javascript的方法,MongoDB可以在服务器端进行js的解析。如果这个方法写的不对,程序将不能正常执行。
 
第二个应用场景:当访问某个具体的内容时,返回某段时间曾经浏览过这篇文章的其他人关注的其他内容,以便对当前用户有一个内容的引导。
  • 首先先找出某段时间内曾经访问该内容的人作为统计的条件,我们使用sessionId而不是userid,是为了将没有登录的用户的访问算进来一起统计
java 代码
  1. DBObject query = newBasicDBObject();   
  2.     query.put("entityId", entityId);   
  3.     query.put("entityName", entityName);   
  4.     query.put("date", newBasicDBObject("$gte", fromDate));   
  5.     query.put("date", newBasicDBObject("$lt", toDate));   
  6. List sessionIds = this.mongoService.getCollection().distinct("sessionId", query);//这里运用了取出结果集中的重复值的函数distinct(String key,DBObject query),相当于SQL:select distinct(name) from table   
  • 定义map方法,主要是定义outputCollection的结构。OutputCollection的输出结构为:{_id:key,value:次数浏览的次数}
java 代码
  1. String mapfun = " function(){emit({url:this.url,title:this.title},1)}";//key={url:this.url,title:this.title},value=reduce方法的返回值。以为是计算数据的次数,所以这里的value定义的是常量1   
  • 定义reduce方法
java 代码
  1. String reducefun = " function(key,vals){var count=0; for(var i in vals){count+=vals[i];} return count;}";//如果同一个key的数据出现的次数进行求和。   
  • 执行
java 代码
  1. *.getCollection().mapReduce(mapfun, reducefun,"pageview_results"new BasicDBObject("sessionId",new BasicDBObject("$in",sessionIds.toArray())));   
  • 遍历pageview_results集合的结果:[{_id:{url:”/blog/yongtree/258”,title:’博客1’},value:’45.0’},{_id:{url:”/blog/slx/288”,title:’博客2’}, value:’30.0’}]
 
前台展现的效果:
  • 当进入用户空间时,展现空间主人关心的其他内容

  • 当浏览某篇内容时,展现其他浏览者的关心的内容

 
继续关注OECP社区,我们将会实践和发布更多基于MongoDB的应用。本着共享的精神,该文档可以被转载和应用,但是要注明出处。
作者主页:http://www.po-soft.com/hi/yongtree
 

附件:


    评分: 请先登录再投票,同一篇博客一月只能投票一次!
    无人投票

相关博客:


评论

谭明智 2010-11-16 9:50:36   回复

特别补充:
最近mongodb频繁的报:error: hashtable namespace index max chain n:26715,才开始以为硬盘空间小了,文件不能创建,但是当扩了硬盘空间后,依然不新增数据文件,还是频繁的报错,最后尝试性的用show collections命令看了一下某个DB下的表,结果让我大吃一惊,该DB下居然有几万个collection,这就是导致服务报错的元凶。
在mapreduce的过程中,mongodb会创建临时collection,用时间戳生成的collection名字。官方介绍,默认的这些临时collection会在链接断开的时候drop掉的,但是现在没有drop掉。于是增加了显式的drop。

java 代码
  1. MapReduceOutput out = this.mongoService.getCollection().mapReduce(mapfun, reducefun,   
  2.                     "page_view_result", query);   
  3.             List<DBObject> list = out.results().limit(limit).sort(   
  4.                     new BasicDBObject("value", -1)).toArray();   
  5.                         out.drop();  


但是依然不行,mongodb的控制台显示:
Tue Nov 16 08:56:36 CMD: drop LogDB.tmp.mr.mapreduce_1289868996_4_inc
Tue Nov 16 08:56:36 CMD: drop LogDB.page_view_result
发现mongodb删除的是指定输出的collection,而对临时创建的collection不管不问。(人为指定临时collection,系统也是自动创建一个时间戳命名的临时collection,最后把这个临时表输出到指定的临时collection上)
最后不再指定临时collection了,传递的outputCollection这个参数设为null,结果控制台上显示:
CMD: drop LogDB.tmp.mr.mapreduce_1289868996_4_inc
CMD: drop LogDB.tmp.mr.mapreduce_1289868996_4
用show collections查看所有的collection,发现已经不再累积临时collection了。

我非常诧异,为什么指定临时collection后,自动创建的collection不drop,是哪里写的不对,还是提供的java驱动做的并不是很完美,希望mongodb能尽快的发展java相关的驱动,也希望mongodb相关的文档能够不断的繁荣起来。

游客 2010-11-20 20:04:10  
支持下谭老大的工作,非常不错的分享

发表评论

关注此文的人们还关注