[Mongo]分组统计时间aggregate，group，distinct
开发中有些按日期记录的记录需要各种维度的统计，按天，按月，按年，按小时，。。分组统计，还有些需要对字段去重统计，在之前的 [Mongo] 按时间分组统计（group时间格式化） 中用group实现了按天的统计，不过使用new Date（）方法会有些坑，今天看了下aggregate中，使用聚合来写个时间统计。
tips: aggregate 挺复杂，弄明白了再做笔记，现在只是根据需求来查询。
数据结构还是：
&_id& : ObjectId(&541fcc51c6c3cd&),
&url& : &/&,
&addtime& : ISODate(&T00:15:02Z&)
&_id& : ObjectId(&541fcc51c6c3ce&),
&url& : &http://meiwen.me/src/index.&,
&addtime& : ISODate(&T00:15:07Z&)
按月统计pv值（相当于group）
db.msds_accessrecord.aggregate([
{$group: {
&$month&: &$addtime&
pv: {$sum: 1}}
{$sort: {&_id&: 1}}
&result& : [
&_id& : 8,
&pv& : 583
&_id& : 9,
&pv& : 1399
按月统计url值，重复url去掉，这里只是做个演示，可能统计没什么意义 （相当于group+distinct)
db.msds_accessrecord.aggregate([
{$group: {
&month&: {&$month&: &$addtime&},
&url&: &$url&
{$group: {_id:&$_id.month&, uv: {$sum: 1}}},
{$sort: {&_id&:1}}
这里用到了管道，排序，聚合
&result& : [
&_id& : 8,
&_id& : 9,
&uv& : 134
聚合使用方法： http://docs.mongodb.org/manual/reference/method/db.collection.aggregate/#db.collection.aggregate
日期聚合函数： http://docs.mongodb.org/manual/reference/operator/aggregation-date/
本文出自 &orangleliu笔记本& 博客，请务必保留此出处http://blog.csdn.net/orangleliu/article/details/创建日期: 16:51:20
1.1连单台mongodb&
Mongo&mg&=&newMongo();//默认连本机127.0.0.1&&端口为27017
Mongo&mg&=&newMongo(ip);//可以指定ip&端口默认为27017
Mongo&mg&=&newMongo(ip,port);//也可以指定ip及端口号
1.2连双台mongodb&
//ip为主机ip地址，port为端口号，dataBaseName相当于数据库名
DBAddress&left&=&new&DBAddress("ip:port/dataBaseName");
&DBAddress&right&=&new&DBAddress("ip:port/dataBaseName");
//若一个mongodb出现问题，会自动连另外一台
Mongo&mongo&=&new&Mongo(left,&right);
1.3连多台mongodb&
List&ServerAddress&&mongoHostList&=&newArrayList&ServerAddress&();
mongoHostList.add(newServerAddress("ip",port));
mongoHostList.add(newServerAddress("ip",port));
mongoHostList.add(newServerAddress("ip",port));
Mongo&mg&=&newMongo(mongoHostList);
2.1获取mongodb的db
dataBaseName相当于关系数据库里的数据库名，mongodb中若没有该数据库名也不会报错，默认mongodb会建立这个数据库名，为空。
DB&db&=&mg.getDB(dataBaseName);&//注意：mongodb区分大小写，程序中一定要注意
2.2mongodb的db安全认证
若这个时候要想访问db中的collection（相当于关系数据库里的表），就必须通过安全认证才可以访问，否则后台会报您没有通过安全认证。&
//安全认证java代码&返回true表示通过，false表示没通过不能进行操作
boolean&auth&=db.authenticate("userName",&"password".toCharArray());
3.1得到mongodb中的db的collection
&DBCollection&users&=&db.getCollection(tableName);
&参数tableName相当于关系数据库里的表名,若mongodb中没有该tableName，默认会创建该tableName,为空&
3.2mongodb中的db的collection自增长主键&
Mongodb中也像传统的关系数据库里表一样，有主键（_id）概念，用来唯一标识他们。当用户往collection中插入一条新记录的时候，
如果没有指定_id属性，那么mongodb会自动生成一个ObjectId类型的值，保存为_id的值。&
3.3java对collection进行插入操作&
//第一种插入方式
DBObject&data1&=&newBasicDBObject();
data1.put("cust_Id",&"123456");
data1.put("is_Show",&1);
data1.put("start_time",&newDate());
users.insert(data1);//等同于users.save(data1);&
//第二中插入方式
BasicDBObjectBuilderdata1&=BasicDBObjectBuilder.start().add(...).add(...);&
users.insert(data1.get());//等同于users.save(data1.get());&
//第三种插入方式
Map&String,&Object&&data1&=&new&HashMap&String,&Object&();
data1.put(...);
users.insert(new&BasicDBObject(data1));//等同于users.save(new&BasicDBObject(data1)
&//第四中插入方式
String&json&=&"{'database':&'mkyongDB','table'&:&'hosting',"&+&&"'detail'&:&{'records'&:&99,&'index'&:&'vps_index1',&'active'&:'true'}}}";
DBObject&data1&=&(DBObject)JSON.parse(json);
users.insert(data1);//等同于users.save(data1);&
3.4java对collection的查询操作
====================================
DBObject&fields&=&new&BasicDBObject();
fields.put("_id",&false);
fields.put("tag",&true);
fields.put("tv_sec",&true);
DBCursor&cur&=&users.find(query,&fields);
可以用来过滤不需要获取的字段，&减少IO
====================================
//查询所有的数据find()&
DBCursor&cur&=&users.find();&while(cur.hasNext()){...}&
//查询id大于等于1的记录，并且只取10条记录
DBCursor&cur&=&users.find(newBasicDBObject("_id",newBasicDBObject("$gte",1))).limit(10);
//查询id大于等于1的记录，并按id进行降序-1表示降序，1升序。
DBCursor&cur&=&users.find(new&BasicDBObject("_id",newBasicDBObject("$gte",1))).sort(newBasicDBObject("_id",-1));&
//查询id大于等于1的记录，并跳过前10条记录显示&相当于id&10
DBCursor&cur&=&users.find(newBasicDBObject("_id",newBasicDBObject("$gte",1))).skip(10);&
//查询id大于等于1的记录，跳过前10条记录并且只显示10条记录。相当//于分页功能where&id&10&and&id&=20
DBCursor&cur&=&users.find(newBasicDBObject("_id",&newBasicDBObject("$gte",1))).skip(10).limit(10);
//查询id大于等于1的所有记录数&返回int型
users.find(newBasicDBObject("_id",newBasicDBObject("$gte",1))).count()
&//findAndRemove()&查询_id=30000的数据，并且删除
users.findAndRemove(newBasicDBObject("_id",&30000));
3.5java对collection的更新操作&
查询id为300的记录，将cust_Id的值更新为6533615，一定要注意大小写，以及数据//类型，返回值为int表示所影响的记录条数可以用users.findOne(newBasicDBObject("_id",300));查看下，会发现这条记录//只返回两个字段了，分别为_id,cust_Id，别的字段都删除了。
users.update(newBasicDBObject("_id",300),&newBasicDBObject&("cust_Id","6533615")).getN();
&//这种写法可以实现只更新id为300的cust_Id值为6533615，而不删除这条记录的别的字//段属性
users.update(newBasicDBObject("_id",300),&newBasicDBObject("$set",newBasicDBObject("cust_Id","6533615"))).getN();
3.6java对collection的删除操作
//移除cust_Id为6533615的数据。注意&用remove方法不释放磁盘空间，
//mongodb只在collection中做了标志，没有正在删除。
users.remove(newBasicDBObject("cust_Id","6533615")).getN();
&//移除id&=1的数据
users.remove(newBasicDBObject("_id",new&BasicDBObject("$gte",1))).getN();
//移除整个collection，drop也不释放磁盘空间
users.drop();
本周实验性地使用上mongodb，应用场景很简单，所以现在对mongodb了解也不是很深入。本文主要介绍mongodb的java客户端编程，这方面的内容也很简单，这里只是做个总结。不得不说，像mongodb这种介于kv和sql之间的存储，对很多的互联网应用很合适。mongodb现在的应用案例已经很多，并且社区的活跃度很高（国内也有不少人对其有很深的研究，如果有时间和精力，或许我也会投入一些对mongodb的研究），很值得期待。
言归正传，下面总结下使用Java开发mongodb应用的一些点滴。在Java中和mongodb交互的最直接的选择就是使用MongoDB&Java&Driver，其下载地址是：/mongodb/mongo-java-driver/downloads。总的来说，在Java中操作mongodb的API还是很简洁，下面对其一些常见的使用做些介绍。
1、连接数据库&
和mongodb建立连接的示例代码如下：
Mongo&m&=new&Mongo("localhost",27017);&DB&db&=&m.getDB("db_test");
尽管这里获得了表示mongodb的db_test数据库连接的对象db，但这时并没有真正和mongodb建立连接，所以即便这时数据库没起来也不会抛出异常，尽管你还是需要catch它的实例化过程。mongodb的java&driver对连接做了池化处理，所以应用中只需要实例化一个Mongo对象即可，对它的操作是线程安全的，这对开发使用来说真的是很方便。
2、取得DBCollection&
mongodb中的collection在Java中使用DBCollection表示（这是一个抽象类，尽管你不必需要知道），创建DBCollection实例也是一行代码，和创建DB实例一样，这个操作并不涉及真正的和数据库之间的通信。
DBCollection&coll&=&db.getCollection("collection1");&
要获得类似mysql中&show&tables&功能，可以使用如下代码：
Set&String&&colls&=&db.getCollectionNames();for(String&s&:&colls){System.out.println(s);}3、插入文档&
mongodb存储JSON格式的文档，而在Java中表示这种数据格式的最简便的类就是Map了。MongoDB&Java&Driver中提供的BasicDBObject就是个Map（它继承自LinkedHashMap并实现DBObject接口），它会将Map中的数据转换成BSON格式传输到mongodb。下面是插入文档的示例：
&&&&&&&&&&DBCollection&coll=db.getCollection("collection1");
&&&&&&&&&&BasicDBObject&doc=new&BasicDBObject();
&&&&&&&&&&doc.put("name","kafka0102");doc.put("age",28);doc.put("time",&newDate());
&&&&&&&&&&coll.insert(doc);
mongodb中每个插入的文档会产生个唯一标识_id。当调用coll.insert(doc);时，driver会检查其中是否有_id字段，如果没有则自动生成ObjectId实例来作为_id的值，这个ObjectId由4部分编码而成：当前时间、机器标识、进程号和自增的整数。insert函数也支持插入文档列表：
&&&&&&&&&&insert(List&DBObject&&list)&
而提交操作也有update(&DBObject&q&,&DBObject&o&)、remove(&DBObject&o&)。
4、查询文档4.1、findOne&
findOne是查询满足条件的第一条记录（不意味着数据库满足条件的只有一条记录），查询条件使用DBObject表示，示例如下：
&&&&&&&&&&DBCollection&coll=db.getCollection("collection1");
&&&&&&&&&&BasicDBObject&cond=new&BasicDBObject();
&&&&&&&&&&cond.put("name","kafka0102");cond.put("age",28);
&&&&&&&&&&DBObject&ret=coll.findOne(cond);System.out.println(ret);
返回结果是个DBObject，可以通过get(key)来取值。对于查询条件，可以通过嵌套多层来表示复杂的格式，比如：
&&&&&&&&&&query=newBasicDBObject();&
&&&&&&&&&&query.put("i",&new&BasicDBObject("$gt",50));//&e.g.&find&all&where&i&&&50
4.2、find&
find函数是查询集合的，它返回的DBCursor是DBObject的迭代器，使用示例如下：
&&&&&&&&&&DBCollection&coll=db.getCollection("collection1");
&&&&&&&&&&BasicDBObject&cond=new&BasicDBObject();cond.put("i",&new&BasicDBObject("$gt",20).append("$lte",30));
&&&&&&&&&&DBCursor&ret=coll.find(cond);
&&&&&&&&&&while(ret.hasNext()){System.out.println(ret.next());}
5、使用索引&
创建索引语句如：coll.createIndex(new&BasicDBObject(&i&,&1));&，其中i表示要索引的字段，1表示升序（-1表示降序）。可以看到，DBObject成为java客户端通用的结构表示。查看索引使用DBCollection.getIndexInfo()函数。
6、MongoDB&Java&Driver的并发性&
前面提到，Java&MongoDB&Driver使用了连接的池化处理，这个连接池默认是保持10个连接，可以通过Option进行修改，在应用中使用Mongo的一个实例即可。连接池中的每个连接使用DBPort结构表示（而不是DBCollection），并寄存于DBPortPool中，所以对DBCollection的操作并不意味着使用同一个连接。如果在应用的一次请求过程中，需要保证使用同一个连接，可以使用下面的代码片断：
DB&db...;&db.requestStart();//code....&db.requestDone();&
在requestStart和requestDone之间使用的连接就不是来自于DBPortPool，而是当前线程中的ThreadLocal结构变量（MyPort中保持了DBPort成员）。
7、其他选择&
尽管Java&mongodb&driver很不错，但就像很多人不使用JDBC而使用一些ORM框架，mongodb的java客户端也有其他的选择。1）对POJO和DAO的支持。对于那些热衷ORM的人来说，Morphia（/p/morphia/wiki/QuickStart）是个不错的选择，它通过在POJO中添加注释来实现映射，并提供对DAO的CRUD操作的支持。2）对DSL的支持。Sculptor就是这样的东西，使用者编写中立的DSL文件，Sculptor将其翻译成代码。这通常不具有吸引力，除非是多语言的应用，能将DSL翻译成多种编程语言，否则除了增加学习成本，没什么收益。3）对JDBC的支持。mongo-jdbc是这样的东西，但现在还是实验性质的。它或许是想亲近Java程序员，不过它显然不能完全兼容JDBC，而很多Java程序员对JDBC也并不感冒，所以它不是很值得使用。
Navigation
李岩的个人博客，记录工作和生活总遇到的问题并付解决方法，也为大家提供方便。
评论最多的文章
访问最多的文章下次自动登录
现在的位置:
& 综合 & 正文
java：Mongodb
分组统计应用group的使用（含场景）
------------------------场景-----------------
mongodb中数据集合（ABC123）包含的指标进行统计分析，指标(指标在mongodb中实际以字母、数字组合的编码表示)包括：
A1：用户标识、A2：用户编码、A3：用户名称、A4：地市、A5：区县、A6：手机号码、A7：账期、A8：入网时间、A9：ARPU值
需地市、区县分组统计的指标(统计指标的结果标识编码可自定义)：
X1：用户数、X2：总ARPU值、X3：平均用户入网时间
即需要统计分析：
count(用户编码)
sum(ARPU值)
平均入网时间：sum(入网时间)/count(用户编码)
----------------------------------------------
mongo-2.7.3.jar
mongodb group 分组函数使用：
BasicDBList basicDBList = (BasicDBList)db.getCollection("mongodb中集合编码或者编码")
.group(DBObject key,
--分组字段，即group by的字段
DBObject cond,
--查询中where条件
DBObject initial,
--初始化各字段的值
String reduce,
--每个分组都需要执行的Function
String finial
--终结Funciton对结果进行最终的处理
------------------------------------------------------
将分组字段、统计的结果字段、统计公式中用到的字段分别保存在HashMap中：
分组字段：
HashMap dimMap = new HashMap();
dimMap.put("A4","A4");
dimMap.put("A5","A5");
统计的结果字段:
HashMap forIdxMap = new HashMap();
dimMap.put("X1","count(A2)");
dimMap.put("X2","sum(A9)");
//总ARPU值
dimMap.put("X3","sum(A8)/count(A2)");
//平均用户入网时间
统计公式中用到的字段:
HashMap indexMap = new HashMap();
dimMap.put("A2","A2");
//用户编码
dimMap.put("A8","A8");
//入网时间
dimMap.put("A9","A9");
java调用：
BasicDBList basicDBList = (BasicDBList)db.getCollection("ABC123")
.group(GroupUtil.generateFormulaKeyObject(dimMap),
new BasicDBObject(),
GroupUtil.generateFormulaInitObject(indexMap),
GroupUtil.generateFormulaReduceObject(indexMap),
GroupUtil.generateFormulaFinalizeObject(forIdxMap, indexMap));
GroupUtil.java：
package com.
import java.util.HashM
import java.util.I
import com.mongodb.BasicDBO
import com.mongodb.DBO
public class GroupUtil
* 方法描述：根据用户选择的维度编码和指标编码，生成Group中的key
* @param dimMap 维度编码
* @return key 对象
public static DBObject generateFormulaKeyObject(HashMap dimMap)
DBObject key = new BasicDBObject();
Iterator dimIt = dimMap.keySet().iterator();
while (dimIt.hasNext())
String dimId = (String)dimIt.next();
key.put(dimId, true);
* 方法描述：根据用户选择的维度编码和指标编码，生成Group中的属性初始化值
* @param dimMap 维度编码
* @return key 对象
public static DBObject generateFormulaInitObject(HashMap indexMap)
DBObject initial = new BasicDBObject();
//设置计算指标中使用的指标对应的统计值：sum、count、avg、max、min
Iterator indexIt = indexMap.keySet().iterator();
while (indexIt.hasNext())
DBObject index = new BasicDBObject();
index.put("count", 0);
index.put("sum", 0);
index.put("max", 0);
index.put("min", 0);
index.put("avg", 0);
index.put("self", 0);
String indexId = (String)indexIt.next();
initial.put(indexId, index);
* 方法描述：根据用户选择的指标编码，生成Group中的reduce函数
* @param indexMap 指标编码
* @return reduce函数
public static String generateFormulaReduceObject(HashMap indexMap)
StringBuffer reduceBuf = new StringBuffer();
reduceBuf.append("function(doc, prev) {");
reduceBuf.append("var tempV");
Iterator indexIt = indexMap.keySet().iterator();
while (indexIt.hasNext())
String indexId = (String)indexIt.next();
//计算指标数量
reduceBuf.append("prev.").append(indexId).append(".count ++;");
//计算指标总计
reduceBuf.append("if(isNaN(").append("prev.").append(indexId).append(".sum").append(")){");
reduceBuf.append("prev.").append(indexId).append(".sum = 0;");
reduceBuf.append("}");
reduceBuf.append("prev.").append(indexId).append(".sum += parseFloat(doc.").append(indexId).append(");");
reduceBuf.append("if(isNaN(").append("prev.").append(indexId).append(".self").append(")){");
reduceBuf.append("prev.").append(indexId).append(".self = 0;");
reduceBuf.append("}");
reduceBuf.append("prev.").append(indexId).append(".self = parseFloat(doc.").append(indexId).append(");");
reduceBuf.append("print(prev.").append(indexId).append(".self);");
//计算指标最大值
reduceBuf.append("tempVal = parseFloat(doc.").append(indexId).append(");");
reduceBuf.append("if(").append("prev.").append(indexId).append(".max == 0").append("){");
reduceBuf.append("prev.").append(indexId).append(".max = tempV");
reduceBuf.append("}else{");
reduceBuf.append("prev.").append(indexId).append(".max = ");
reduceBuf.append("prev.").append(indexId).append(".max & tempVal ? ");
reduceBuf.append("prev.").append(indexId).append(".max : tempV");
reduceBuf.append("}");
//计算指标最小值
reduceBuf.append("if(").append("prev.").append(indexId).append(".min == 0").append("){");
reduceBuf.append("prev.").append(indexId).append(".min = tempV");
reduceBuf.append("}else{");
reduceBuf.append("prev.").append(indexId).append(".min = ");
reduceBuf.append("prev.").append(indexId).append(".min & tempVal ? ");
reduceBuf.append("prev.").append(indexId).append(".min : tempV");
reduceBuf.append("}");
//计算指标的平均值
reduceBuf.append("prev.").append(indexId).append(".avg = ");
reduceBuf.append("prev.").append(indexId).append(".sum");
reduceBuf.append(" / ");
reduceBuf.append("prev.").append(indexId).append(".");
reduceBuf.append("}");
return reduceBuf.toString();
* 方法描述：根据用户选择的指标编码，生成MapReduce中的finalize函数
* @param indexMap 指标编码
* @return reduce函数
public static String generateFormulaFinalizeObject(HashMap forIdxMap, HashMap indexMap)
StringBuffer reduceBuf = new StringBuffer();
reduceBuf.append("function(doc){");
//得到计算指标的公式运行值
Iterator formulaIt = forIdxMap.keySet().iterator();
while (formulaIt.hasNext())
String indexId = (String)formulaIt.next();
String idxFormula = (String)forIdxMap.get(indexId);
reduceBuf.append("var tempIdx, tempF");
Iterator indexItB = indexMap.keySet().iterator();
int i = 0;
while (indexItB.hasNext())
String indexIdS = (String)indexItB.next();
if(i == 0)
reduceBuf.append("tempFormula = \"").append(idxFormula).append("\";");
reduceBuf.append("tempIdx = ").append("doc.").append(indexIdS).append(".");
reduceBuf.append("tempFormula = ").append("tempFormula").append(".replace(/sum\\(").append(indexIdS).append("\\)/g,tempIdx);");
reduceBuf.append("tempIdx = ").append("doc.").append(indexIdS).append(".");
reduceBuf.append("tempFormula = ").append("tempFormula").append(".replace(/count\\(").append(indexIdS).append("\\)/g,tempIdx);");
reduceBuf.append("tempIdx = ").append("doc.").append(indexIdS).append(".");
reduceBuf.append("tempFormula = ").append("tempFormula").append(".replace(/min\\(").append(indexIdS).append("\\)/g,tempIdx);");
reduceBuf.append("tempIdx = ").append("doc.").append(indexIdS).append(".");
reduceBuf.append("tempFormula = ").append("tempFormula").append(".replace(/max\\(").append(indexIdS).append("\\)/g,tempIdx);");
reduceBuf.append("tempIdx = ").append("doc.").append(indexIdS).append(".");
reduceBuf.append("tempFormula = ").append("tempFormula").append(".replace(/avg\\(").append(indexIdS).append("\\)/g,tempIdx);");
reduceBuf.append("var resTemp = ").append("eval(tempFormula);");
reduceBuf.append("doc.").append(indexId).append(" = resTemp.toFixed(2);");
Iterator indexItC = indexMap.keySet().iterator();
while (indexItC.hasNext())
String indexId = (String)indexItC.next();
reduceBuf.append("doc.").append(indexId).append(" = doc.").append(indexId).append(".");
reduceBuf.append("}");
return reduceBuf.toString();
注：group限制：分组后的组数不得超过20000个组！
&&&&推荐文章:
【上篇】【下篇】mongodb-java-driver基本用法_百度文库
两大类热门资源免费畅读
续费一年阅读会员，立省24元！
mongodb-java-driver基本用法
上传于||暂无简介
阅读已结束，如果下载本文需要使用1下载券
想免费下载本文？
下载文档到电脑，查找使用更方便
还剩3页未读，继续阅读
你可能喜欢<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN"
您的访问请求被拒绝 403 Forbidden - ITeye技术社区
您的访问请求被拒绝
亲爱的会员，您的IP地址所在网段被ITeye拒绝服务，这可能是以下两种情况导致：
一、您所在的网段内有网络爬虫大量抓取ITeye网页，为保证其他人流畅的访问ITeye，该网段被ITeye拒绝
二、您通过某个代理服务器访问ITeye网站，该代理服务器被网络爬虫利用，大量抓取ITeye网页
请您点击按钮解除封锁&