本文共 4013 字，大约阅读时间需要 13 分钟。
MapReduce是聚合工具中的明星。Count、distinct、group能做的上述事情Mapreduce都能做。他是一个可以轻松并行化到多个服务器的聚合方法。它会拆分问题，再将各个部分发送到不同的机器上，让每台机器都完成一部分。当所有机器都完成的时候，再把结果汇集起来形成最终完整的结果。
MapReduce需要几个步骤。最开始是映射（map），将操作映射到集合中的每个文档。这个操作要么“无作为”，要么“产生一些键和X个值”。然后就是中间环节，称作洗牌（shuffle），按照键分组，并将产生的键值组成列表放到对应的键中。化简（reduce）则把列表中的值化简成一个单值。这个值被返回，然后接着进行洗牌，直到每个键的列表只有一个值为止，这个值也就是最终结果。
使用mapreduce的代价就是速度：group不是很快，mapreduce更慢，绝不要用在“实时”环境中。要作为后台任务来运行MapReduce，将创建一个保存结果的集合，可以对这个集合进行实时查询。
对消息体统的一个统计：
| public List<MsgAnalysisByDate> mapreduceAndroidPushedByDate(Date timeBegin, Date timeEnd) { log.info("--------------安卓推送统计开始mapreduceAndroidPushedByDate-----begin"); MsgMapReduceBo msgMapReduceBo = new MsgMapReduceBo(timeBegin, timeEnd, bcandroidpushrecordMongoDBClient, MsgFlagTypeEnum.PUSHED.getCode(), msgIdName, titleName); return doMapReduce(msgMapReduceBo); } private List<MsgAnalysisByDate> doMapReduce(MsgMapReduceBo msgMapReduceBo) { int type = AppTypeEnum.JD.getCode(); //掌上京东 String outputTarget = livenessMapreduce.buildOutputTarget("tmp_liveness_d_" + type); List<MsgAnalysisByDate> result = new ArrayList<MsgAnalysisByDate>(200); String map = buildMapForBC(msgMapReduceBo.getMsgIdName(), msgMapReduceBo.getTitleName()); String reduce = buildReduceForBC(msgMapReduceBo.getMsgIdName()); QueryBuilder query1 = new QueryBuilder(); query1.and("created").greaterThanEquals(msgMapReduceBo.getTimeBegin()).lessThan(msgMapReduceBo.getTimeEnd());//统计安装量，用created DBCollection collection = msgMapReduceBo.getMongoDBClient().mapReduce(map, reduce, outputTarget, query1).getOutputCollection(); log.info("domapReduce结束：客户端："+msgMapReduceBo.getMongoDBClient()); DBCursor cursor = collection.find(); while (cursor.hasNext()) { DBObject item = cursor.next(); //如果主键为空,忽略记录 DBObject value = (DBObject) item.get("value"); Double count = (Double) value.get("count") == null ? 0 : (Double) value.get("count");//总量 String title = (String) value.get("title") == null ? "" : (String) value.get("title"); String msgId = (String) value.get(msgMapReduceBo.getMsgIdName()); String platformReal = (String) value.get("platform") == null ? "" : (String) value.get("platform"); if (value.toMap().size() == 2) { //如果返回的value尺寸为2，说明只有一条记录，因为只有一条记录时没有进行reduce，结果集是map的， String _id = (String) item.get("_id"); int index = _id.indexOf("-"); msgId = _id.substring(0, index); int idex2 = msgId.indexOf("_"); msgId = msgId.substring(idex2 + 1, msgId.length()); platformReal = _id.substring(index + 1, _id.length()); } MsgAnalysisByDate tmp = new MsgAnalysisByDate(); tmp.setMsgid(msgId); tmp.setTile(title); tmp.setType(type); tmp.setFlag(msgMapReduceBo.getFlag()); if (PlatformEnum.getByValue(platformReal) != null) { tmp.setPlatform(PlatformEnum.getByValue(platformReal).getKey()); } else { log.error("msgId=" + msgId + "platformReal=" + platformReal); } if (msgMapReduceBo.isRead()) { tmp.setReadAmount(count.intValue()); } else { tmp.setAmount(count.intValue()); } result.add(tmp); } collection.drop(); return result; } /** * 注意这里的key，用横杠分割，用于处理多个goupby 字段 * * @return */ private String buildMapForBC(String id, String title) { return "function()" + "{" + "var key1=this." + id + ";" + "var prefix=\'-\';" + "var key2=this.platform;" + "emit(key1+prefix+key2,{title:this." + title + ",count:1});" + "};"; } /** * 注意冗余title * * @return */ public String buildReduceForBC(String id) { return "function(key,values)" + "{" + "var total = 0;" + "var title=values[0].title;" + " var locate = key.indexOf(\'-\');" + " var platform = key.indexOf(\'-\');" + "var platform=values[0].title;" + "for(var i=0;i<values.length;i++){" + "total += values[i].count;" + "}" + "return {" + id + ":key.substr(0,locate) ,title: title, platform:key.substr(locate+1,key.length), count : total};" + "};"; } |
说明：
1. Map阶段：{ key1+prefix+key2}作为key，{title:this." + title + ",count:1}作为value，遍历每一个doc，生成一堆上述（key，value）对的文件
2. Shuffle阶段：把上述的doc文件。Value合并，变成（key，values{value1,value2...valuen}）;
3. Reduce阶段：把上述的结果文件进行reduce处理，具体逻辑就是reduce函数，本例中就是遍历values，取count，汇总总数，值得注意的是这里用了一个技巧：key用了多个字段拼接，用于返回多个值。
转载地址：http://lzrrb.baihongyu.com/