背景
最近做了几个规则逻辑。用到mongo查询比较多，就是查询交易信息跑既定规则筛选出交易商户，使用聚合管道进行统计和取出简单处理后的数据，用SQL代替业务代码逻辑的判断。
方法
MongoDB聚合使用aggregate，聚合管道采取自动向下子执行方式，基本语法格式：
db.COLLECTION_NAME.aggregate(AGGREGATE_OPERATION)
聚合框架中常用的操作:
- $project：修改输入文档的结构。可以用来重命名、增加或删除域，也可以用于创建计算结果以及嵌套文档。
- (match：用于过滤数据，只输出符合条件的文档。)match使用MongoDB的标准查询操作。
- $limit：用来限制MongoDB聚合管道返回的文档数。
- $skip：在聚合管道中跳过指定数量的文档，并返回余下的文档。
- $unwind：将文档中的某一个数组类型字段拆分成多条，每条包含数组中的一个值。
- $group：将集合中的文档分组，可用于统计结果。
- $sort：将输入文档排序后输出。
- $geoNear：输出接近某一地理位置的有序文档。
| 表达式 | 描述 | 实例 |
|---|---|---|
| $sum | 计算总和。 | db.mycol.aggregate([{(group : {_id : ")by_user", num_tutorial : {(sum : ")likes"}}}]) |
| $avg | 计算平均值 | db.mycol.aggregate([{(group : {_id : ")by_user", num_tutorial : {(avg : ")likes"}}}]) |
| $min | 获取集合中所有文档对应值得最小值。 | db.mycol.aggregate([{(group : {_id : ")by_user", num_tutorial : {(min : ")likes"}}}]) |
| $max | 获取集合中所有文档对应值得最大值。 | db.mycol.aggregate([{(group : {_id : ")by_user", num_tutorial : {(max : ")likes"}}}]) |
| $push | 在结果文档中插入值到一个数组中。 | db.mycol.aggregate([{(group : {_id : ")by_user", url : {(push: ")url"}}}]) |
| $addToSet | 在结果文档中插入值到一个数组中，但不创建副本。 | db.mycol.aggregate([{(group : {_id : ")by_user", url : {(addToSet : ")url"}}}]) |
| $first | 根据资源文档的排序获取第一个文档数据。 | db.mycol.aggregate([{(group : {_id : ")by_user", first_url : {(first : ")url"}}}]) |
| $last | 根据资源文档的排序获取最后一个文档数据 | db.mycol.aggregate([{(group : {_id : ")by_user", last_url : {(last : ")url"}}}]) |
查询示例
示例一
部分字段说明：transAmt:交易金额，transType:交易类型，transTime:交易时间，mercNum:商户编号
查询交易信息，交易商户昨天交易笔数大于三百，交易金额累加大于三百万，这里现根据$match将交易信息筛选出来，然后使用$group根据商户编号分组，统计交易笔数和累加交易金额，将分组结果判断匹配交易笔数大于三百，交易金额大于三百万。
db.getCollection("box_order").aggregate([
{
$match: {
"transTime":{$gte:ISODate("2020-01-03T00:00:00.000Z"),$lt:ISODate("2020-01-10T00:00:00.000Z")},
"transType":"consume",
"transStatus":{$in:["tsProcessing","success"]}
}
},
{
$group: {
"_id": "$mercNum",
"count": {"$sum": 1},
"totalAmt": {"$sum": "$transAmt"}
}
},
{
$match: {
"count": {"$gte": 300},
"totalAmt": {"$gte": 3000000}
}
}
])
示例二
部分字段说明：cardNo:交易卡号，transType:交易类型，transTime:交易时间，mercNum:商户编号
查询时间段内指定卡号下的交易商户信息。
根据卡号和交易时间将交易数据查出来，然后只显示商户号和卡号两列字段，根据商户号和卡号分组去重，再根据卡号分组，将商户号转化成一个字段变成数组。
db.getCollection("order_202011").aggregate([
{
"$match": {
"detailInfo.cardNo": {
"$in": [
"YtCZ7KhCVG5xerKUg8bzJhVAjW/hWAWj",
"cQ7QQ0yCVW6LhHtJNVRq2A==",
"6KDpHmQ9s+0SQAGAUyLJ4A==",
"cQ7QQ0yCVW7iSegn8uqIfg==",
"ZEOcXdI4rfvswAz7dQ80hw==",
"6KDpHmQ9s+2Nz61PPuOamw=="
]
},
"baseInfo.transTime": {
"$gte": new Date(2020,10,01),
"$lt": new Date(2020,10,24)
}
}
},
{
"$project": {
"merchantInfo.mercNum": 1,
"detailInfo.cardNo": 1
}
},
{
"$group": {
"_id": {
"mercNum": "$merchantInfo.mercNum",
"cardNo": "$detailInfo.cardNo"
}
}
},
{
"$group": {
"_id": "$_id.cardNo",
"mercNums": {
"$push": "$_id.mercNum"
}
}
}
])
示例三
根据指定商户和其他条件查询交易信息，根据卡号分组并组装成一个字段的集合，最后筛选掉id只保留cardNos数组
db.getCollection("box_order_fxq_202104").aggregate([
{
"$match": {
"mercNum": "M15201812030753174730",
"transTime": {
"$gte": ISODate("2021-04-17T16:00:00.000Z"),
"$lt": ISODate("2021-04-18T16:00:00.000Z")
},
"mercLevel": {
"$in": [
"C",
"D",
"E"
]
},
"payType": "POSPAY",
"transType": "consume",
"cardType": "2"
}
},
{
"$group": {
"_id": null,
"cardNos": {
"$push": "$cardNo" //$addToSet
}
}
},
{
"$project":{
"cardNos":1,"_id":0
}
}
])
查询结果：
{
"cardNos" : [
"n2IwHHhfEAJcm6RFsoNPcBVAjW/hWAWj",
"n2IwHHhfEAJcm6RFsoNPcBVAjW/hWAWj"
]
}
示例四
根据时间查询交易信息后，根据商户号分组，并将第一个交易信息存放入data字段中。（如果是需要全部的商户交易信息那么将$first修改为$push）
db.getCollection("order").aggregate([
{
"$match": {
"startTrxTime": {
"$gte": ISODate("2021-07-20T16:00:00.000Z"),
"$lt": ISODate("2021-07-21T16:00:00.000Z")
}
}
},
{
"$group": {
"_id": "$subMerchantNo",
"data":{"$first": "$$ROOT"} //$push
}
},
{
"$sort": {
"_id": 1
}
}
])
尾言
最近那个到查询的大差不差，要注意的都是一些小改动，一般情况正常查就可以。后续有什么不一样的会继续补充。先到这里
