Python语言环境中MongoDB聚合经验一则——分组取最近一次记录集

Posted 2021-12-14 肖永威

0. 需求描述

按客户号“carduser_id”分组查询，并按时间“occurtime”排序取最大时间（排序后的第一行记录），也就是取每个客户号下最近一次记录集，而且，此记录集升格为根记录。

1. MongoDB聚合案例

（1）聚合排序“$sort”，需要对应“allowDiskUse:true”参数；
（2）指定文档提升到顶级并替换所有其它字段
描述：将指定文档提升到顶级并替换所有其它字段。该操作会替换输入文档中的所有现在字段，包括_id字段。

操作语法：$replacRoot: newRoot:<replacementDocument>

db.getCollection('customerfeature').aggregate([
'$sort':'occurtime':-1,
'$group':'_id': '$carduser_id',data:$first: '$$ROOT',
'$replaceRoot':newRoot: '$data'],
allowDiskUse:true)

2. pymongo聚合

pymongo函数方法：collection.aggregate(pipeline, session=None, **kwargs)
在此集合上使用聚合框架执行聚合。

其中，对应上面“allowDiskUse:true”参数为：allowDiskUse=True

import pymongo
import pandas as pd
import json

def exportdata():
    db_name = 'ccm'
    collection_name = 'trainfeature'
    pipeline  = ['$sort':'occurtime':-1,
                    '$group':'_id': '$carduser_id','data':'$first': '$$ROOT',
                    '$replaceRoot':'newRoot': '$data']
    client = pymongo.MongoClient('mongodb://dba:1***@192.168.*.*:27017/ccm')
    db = client[db_name]
    # 获取原集合
    collection = db[collection_name]     
    df = pd.DataFrame(list(collection.aggregate(pipeline,allowDiskUse=True)))
    del df['_id']
    print('数据集行数',len(df))
    #print(df)
    # 新建特征集合
    collection_new = db['customerfeature']     
    try:      
        collection_new.insert(json.loads(df.T.to_json()).values())
    except:
        print(Exception)
        print(df)   

if __name__ == '__main__':
    exportdata()
    pass