Monogdb使用 MapReduce进行分组统计查询

1/**
2
3  * 
4
5  * @param businessNo 
6
7  * @param beginTime 开始时间
8
9  * @param endTime 结束时间
10
11  * @param pageNo  页码
12
13  * @param pageSize 分页大小
14
15  * @return
16
17  */
18
19 public static List<DBObject> query(Integer businessNo,Date beginTime,Date endTime,int pageNo,int pageSize){
20
21
22  StringBuffer map = new StringBuffer();
23
24  map.append("function(){ ");
25
26  map.append("  var t = new Date(this.createTime);");
27
28  map.append("  emit({ ");
29
30  map.append("     orgId:this.orgId,");
31
32  map.append("     operatorId:this.operatorId,");
33
34  map.append("     objId:this.businessParam.objId,");
35
36  map.append("     creatDate:t.getFullYear() +'-'+(t.getMonth() + 1)+'-'+t.getDate()");
37
38  map.append("   },");
39
40  map.append("   {count:1}");
41
42  map.append("  ); ");
43
44  map.append("} ");
45
46  log.info(">>>>>>>map:{}",map.toString());
47
48  
49
50  StringBuffer reduce = new StringBuffer();
51
52  reduce.append("function(key, values) {");
53
54  reduce.append("  var total = 0;");
55
56  reduce.append("  for (var i=0; i<values.length; i++) { ");
57
58  reduce.append("     total += values[i].count;");
59
60  reduce.append("  } ");
61
62  reduce.append("  return {count: total}; ");
63
64  reduce.append("} ");
65
66  log.info(">>>>>>>reduce:{}",reduce.toString());
67
68  
69
70  StringBuffer finalize = new StringBuffer();
71
72  finalize.append("function(key, rvalues) { ");
73
74  finalize.append("   var sum = 0; ");
75
76  finalize.append("   if(rvalues.count>0){ ");
77
78  finalize.append("      rvalues.get='true'; ");
79
80  finalize.append("   }else{ ");
81
82  finalize.append("      rvalues.get='false'");
83
84  finalize.append("   } ");
85
86  finalize.append("   return rvalues; ");
87
88  finalize.append("}");
89
90  log.info(">>>>>>>finalize:{}",finalize.toString());
91
92  
93
94  //查询条件
95
96  DBObject query = new BasicDBObject();
97
98  query.put("businessNo", businessNo);
99
100  query.put("operatorId", new BasicDBObject("$exists", true).append("$ne", 0));
101
102  query.put("createTime", new BasicDBObject("$gte", beginTime.getTime()).append("$lte", endTime.getTime()));
103
104  query.put("businessParam.objId", new BasicDBObject("$exists", true));
105
106 
107  DBObject command = new BasicDBObject();
108
109  command.put("mapreduce", "MyLogs");
110
111  command.put("query", query);
112
113  command.put("map", map.toString());
114
115  command.put("reduce", reduce.toString());
116
117  command.put("out", "MapReduceCommand.OutputType.INLINE");
118
119  command.put("verbose", true);
120
121  
122
123  DBCollection db = MongoDbInit.getDb().getCollection("MyLogs");
124
125  command.put("finalize", finalize.toString());
126
127  MapReduceOutput out = db.mapReduce(command);
128
129  
130
131  DBObject having = new BasicDBObject();
132
133  having.put("value.get", "true");
134
135  
136
137  DBObject orderBy = new BasicDBObject();
138
139  orderBy.put("createTime", -1);//按createTime倒序排  
140
141  
142
143  int skip = (pageNo-1)*pageSize;
144
145  List<DBObject> resultList = new ArrayList();
146
147  resultList = out.getOutputCollection().find(having).sort(orderBy).skip(skip).limit(pageSize).toArray();
148
149  return resultList;
150
151 }

数据结构

1{
2
3  "_id" : ObjectId("52f9c392b7603fe75cff49e7"),
4
5  "deviceId" : "ffffffff-d0b9-867d-191b-921900000000",
6
7  "operationTime" : NumberLong("1392100740000"),
8
9  "operatorType" : 1,
10
11  "operatorId" : 3587,
12
13  "platform" : "android",
14
15  "businessStatus" : 1,
16
17  "sessionId" : "",
18
19  "orgId" : 104,
20
21  "businessName" : "活动详情",
22
23  "createTime" : NumberLong("1392100242676"),
24
25  "businessParam" : {
26
27    "objId" : 653,
28
29    "time" : 715,
30
31    "objType" : 1
32
33  },
34
35  "businessNo" : 1101
36
37}