欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

MongoDB 聚合分组取第一条记录的案例及实现

程序员文章站 2022-11-21 20:35:58
关键字:MongoDB; aggregate;forEach 今天开发同学向我们提了一个紧急的需求,从集合mt_resources_access_log中,根据字段refererDomain分组,取分组中最近一笔插入的数据,然后将这些符合条件的数据导入到集合mt_resources_access_l ......

关键字:mongodb; aggregate;foreach

今天开发同学向我们提了一个紧急的需求,从集合mt_resources_access_log中,根据字段refererdomain分组,取分组中最近一笔插入的数据,然后将这些符合条件的数据导入到集合mt_resources_access_log_new中。

接到这个需求,还是有些心虚的,原因有二,一是,业务需要,时间紧;二是,实现这个功能mongodb聚合感觉有些复杂,聚合要走好多步。

数据记录格式如下:

记录1

{
    "_id" : objectid("5c1e23eaa66bf62c0c390afb"),
    "_class" : "c1",
    "resourceurl" : "/static/js/p.js",
    "refererdomain" : "1234",
    "resourcetype" : "static_resource",
    "ip" : "17.17.13.13",
    "createtime" : isodate("2018-12-22t19:45:46.015+08:00"),
    "disabled" : 0
}

记录2

{
    "_id" : objectid("5c1e23eaa66bf62c0c390afb"),
    "_class" : "c1",
    "resourceurl" : "/static/js/p.js",
    "refererdomain" : "1234",
    "resourcetype" : "dome_resource",
    "ip" : "17.17.13.14",
    "createtime" : isodate("2018-12-21t19:45:46.015+08:00"),
    "disabled" : 0
}

记录3

{
    "_id" : objectid("5c1e23eaa66bf62c0c390afb"),
    "_class" : "c2",
    "resourceurl" : "/static/js/p.js",
    "refererdomain" : "1235",
    "resourcetype" : "static_resource",
    "ip" : "17.17.13.13",
    "createtime" : isodate("2018-12-20t19:45:46.015+08:00"),
    "disabled" : 0
}

记录4

{
    "_id" : objectid("5c1e23eaa66bf62c0c390afb"),
    "_class" : "c2",
    "resourceurl" : "/static/js/p.js",
    "refererdomain" : "1235",
    "resourcetype" : "dome_resource",
    "ip" : "17.17.13.13",
    "createtime" : isodate("2018-12-20t19:45:46.015+08:00"),
    "disabled" : 0
}
 

以上是我们的4条记录,类似的记录文档有1500w。

因为情况特殊,业务发版需要这些数据。催的比较急,而 通过 聚合 框架aggregate,短时间有没有思路, 所以,当时就想着尝试采用其他方案。

最后,问题处理方案如下。

step 1  通过聚合框架 根据条件要求先分组,并将新生成的数据输出到集合mt_resources_access_log20190122 中(共产生95笔数据);

实现代码如下:

db.log_resources_access_collect.aggregate(
                     [
                      
                       { $group: { _id: "$refererdomain" } },
                       { $out : "mt_resources_access_log20190122" }
                     ]
  
                   )

step 2  通过2次 foreach操作,循环处理 mt_resources_access_log20190122和mt_resources_access_log的数据。

代码解释,处理的逻辑为,循环逐笔取出mt_resources_access_log20190122的数据(共95笔),每笔逐行加工处理,处理的逻辑主要是  根据自己的_id字段数据(此字段来自mt_resources_access_log聚合前的refererdomain字段), 去和 mt_resources_access_log的字段 refererdomain比对,查询出符合此条件的数据,并且是按_id 倒序,仅取一笔,最后将join刷选后的数据insert到集合mt_resources_access_log_new。

新集合也是95笔数据。

大家不用担心性能,查询语句在1s内实现了结果查询。

db.mt_resources_access_log20190122.find({}).foreach(
    function(x) {
        db.mt_resources_access_log.find({ "refererdomain": x._id }).sort({ _id: -1 }).limit(1).foreach(
            function(y) {
                db.mt_resources_access_log_new.insert(y)
            }
        )
    }
)

 

step 3 查询验证新产生的集合mt_resources_access_log_new,结果符合业务要求。 

刷选前集合mt_resources_access_log的数据量为1500多w。

刷选后产生新的集合mt_resources_access_log_new 数据量为95笔。

 

注意:根据时间排序的要求,因为部分文档没有createtime字段类型,且 createtime字段上没有创建索引,所以未了符合按时间排序我们采用了sort({_id:1})的变通方法,因为_id 还有时间的意义。下面的内容为mongodb对应_id 的相关知识。

MongoDB 聚合分组取第一条记录的案例及实现

最重要的是前4个字节包含着标准的unix时间戳。后面3个字节是机器id,紧接着是2个字节的进程id。最后3个字节存储的是进程本地计数器。计数器可以保证同一个进程和同一时刻内不会重复。

 

本文版权归作者所有,未经作者同意不得转载,谢谢配合!!!