MongoDB写MapReduce程序

爱被打了一巴掌 2022-05-22 10:37 187阅读 0赞

需求

MongoDB库里存入了短信发送批次信息，如：

![70][]

先需姜红框内的数据进行汇总统计，结果如：

![70 1][]

脚本如下：

db.BatchInfo.mapReduce( 
       function() {emit(((this.t+"").substring(0,8) + this.SmsCnl ),this);}, 
       function(key, values) {
           var touchNum=0;
           var unknow=0;
           var sendFail=0;
           var sendErr=0;
           var sendNum=0;
           var total=0;
           var cancel=0;
           var count=1;
           values.forEach(function(val){ 
               if(val.touchNum != null && val.sts == 1){touchNum += val.touchNum;}
               if(val.sendNum != null && val.sts == 2){cancel += val.sendNum;}
               if(val.unknow != null){unknow += val.unknow;}
               if(val.sendFail != null){sendFail += val.sendFail;}
               if(val.sendNum != null){sendErr = sendErr + (val.sendNum - (val.touchNum == null ?0:val.touchNum) - (val.unknow == null ?0:val.unknow))};
               if(val.sendNum != null){sendNum += val.sendNum;}
               if(val.total != null){total += val.total;}
               count += 1;
               }); 
           return {"smsCnl":key.substring(8),"day":key.substring(0,8),"touchNum":touchNum,"unknow":unknow,"sendFail":sendFail,"sendErr":sendErr,"sendNum":sendNum,"total":total,"count":count,"cancel":cancel};  
           }, 
          {  
             query:{ },  
             out:"sms_report" 
          }
    );

java代码实现：

/**
         * 获取统计报告
         *
         * @param smsCnl 渠道编码
         * @param beginTime 开始时间
         * @param endTime 结束时间
         * @param isp 运营商
         * @param aera 号码归属地
         * @param dayFlag 日期标识
         * @return 统计报告
         */
        @Override
        public Iterator<BasicDBObject> getReport(String smsCnl, Long beginTime, Long endTime, String isp, String aera,String dayFlag) {
            Query query = new Query();
            Criteria criteria = new Criteria() ;
            // 根据发送时间检索
            criteria.and("t").gte(beginTime).lte(endTime);
            if(!Util.isNull(smsCnl)){
            	criteria.and("SmsCnl").is(smsCnl);
            }
            query.addCriteria(criteria) ;
            MapReduceOptions options = MapReduceOptions.options();//设置reduce配置项
            options.outputCollection("SmsReport");
            String mapFunction = "function() {emit(((this.t+\"\").substring(0,8) + this.SmsCnl ),this);}";//map方法编写
            String reduceFunction = "function(key, values) {\n" +
                    "       var touchNum=0;\n" +
                    "       var unknow=0;\n" +
                    "       var sendErr=0;\n" +
                    "       var sendNum=0;\n" +
                    "       var total=0;\n" +
                    "       var cancel=0;\n" +
                    "       var count=0;\n" +
                    "       values.forEach(function(val){ \n" +
                    "           if(val.touchNum != null){touchNum += val.touchNum;}\n" +
                    "           if(val.sendNum != null && val.sts == 2){cancel += val.sendNum;}\n" +
                    "           if(val.sendNum != null){sendErr = sendErr + (val.sendNum - (val.touchNum == null ?0:val.touchNum) - (val.unknow == null ?0:val.unknow))};" +
                    "           if(val.sendNum != null){sendNum += val.sendNum;}\n" +
                    "           if(val.unknow != null){unknow += val.unknow;}\n" +
                    "           if(val.total != null){total += val.total;}\n" +
                    "           count += 1;\n" +
                    "           }); \n" +
                    "       return {\"SmsCnl\":key.substring(8),\"t\":key.substring(0,8),\"touchNum\":touchNum,\"unknow\":unknow,\"sendErr\":sendErr,\"sendNum\":sendNum,\"total\":total,\"count\":count,\"cancel\":cancel}; \n" +
                    "      }"; // 输出字段尽量和原有字段保持一致，不然会出现某条记录只有一条而整个对象打印
            
            MapReduceResults<BasicDBObject> mapReduceResults ;        
            mapReduceResults = this.baseMongoDAO.mapReduce(query,"BatchInfo",mapFunction,reduceFunction,options,BasicDBObject.class);       
            return mapReduceResults.iterator();
        }

[70]: /images/20220522/c075e947e47b4919ab128d40a949137d.png
[70 1]: /images/20220522/7121f79ea54e4270bcff36eb4f2bc049.png