MongoDB Aggregation:从前一行的总和计算Running Totals

示例文档:

{ time: ISODate("2013-10-10T20:55:36Z"), value: 1 }
{ time: ISODate("2013-10-10T22:43:16Z"), value: 2 }
{ time: ISODate("2013-10-11T19:12:66Z"), value: 3 }
{ time: ISODate("2013-10-11T10:15:38Z"), value: 4 }
{ time: ISODate("2013-10-12T04:15:38Z"), value: 5 }

很容易得到按日期分组的聚合结果。但我想要的是查询返回总跑数的结果聚合的，如:

{ time: "2013-10-10" total: 3, runningTotal: 3  }
{ time: "2013-10-11" total: 7, runningTotal: 10 }
{ time: "2013-10-12" total: 5, runningTotal: 15 }

这是可能的MongoDB聚合?

编辑:自MongoDB v5.0以来，首选的方法将是使用Xavier Guihot共享的新的$setWindowFields聚合阶段。

这是你需要的。我对数据中的时间进行了规范化，使它们组合在一起(您可以这样做)。这个想法是$group，并将time和total推入单独的数组。然后$unwind time数组，您已经为每个time文档制作了totals数组的副本。然后可以从包含不同时间的所有数据的数组中计算runningTotal(或类似滚动平均值的东西)。$unwind生成的'index'是time对应的total的数组索引。在$unwind之前使用$sort是很重要的，因为这可以确保数组的顺序是正确的。

db.temp.aggregate(
    [
        {
            '$group': {
                '_id': '$time',
                'total': { '$sum': '$value' }
            }
        },
        {
            '$sort': {
                 '_id': 1
            }
        },
        {
            '$group': {
                '_id': 0,
                'time': { '$push': '$_id' },
                'totals': { '$push': '$total' }
            }
        },
        {
            '$unwind': {
                'path' : '$time',
                'includeArrayIndex' : 'index'
            }
        },
        {
            '$project': {
                '_id': 0,
                'time': { '$dateToString': { 'format': '%Y-%m-%d', 'date': '$time' }  },
                'total': { '$arrayElemAt': [ '$totals', '$index' ] },
                'runningTotal': { '$sum': { '$slice': [ '$totals', { '$add': [ '$index', 1 ] } ] } },
            }
        },
    ]
);

我在一个包含~ 80000个文档的集合上使用了类似的东西，聚合到63个结果。我不确定它在较大集合上的工作效果如何，但我发现，一旦数据减少到可管理的大小，对聚合数据执行转换(投影，数组操作)似乎不会有很大的性能成本。

从Mongo 5开始，对于新的$setWindowFields聚合操作符来说，这是一个完美的用例:

// { time: ISODate("2013-10-10T20:55:36Z"), value: 1 }
// { time: ISODate("2013-10-10T22:43:16Z"), value: 2 }
// { time: ISODate("2013-10-11T12:12:66Z"), value: 3 }
// { time: ISODate("2013-10-11T10:15:38Z"), value: 4 }
// { time: ISODate("2013-10-12T05:15:38Z"), value: 5 }
db.collection.aggregate([
  { $group: {
    _id: { $dateToString: { format: "%Y-%m-%d", date: "$time" } },
    total: { $sum: "$value" }
  }},
  // e.g.: { "_id" : "2013-10-11", "total" : 7 }
  { $set: { "date": "$_id" } }, { $unset: ["_id"] },
  // e.g.: { "date" : "2013-10-11", "total" : 7 }
  { $setWindowFields: {
    sortBy: { date: 1 },
    output: {
      running: {
        $sum: "$total",
        window: { documents: [ "unbounded", "current" ] }
      }
    }
  }}
])
// { date: "2013-10-11", total: 7, running: 7 }
// { date: "2013-10-10", total: 3, running: 10 }
// { date: "2013-10-12", total: 5, running: 15 }

让我们关注$setWindowFields阶段:

按时间顺序$sort s按日期分组的文件:sortBy: { date: 1 }
在每个文档中添加running字段(output: { running: { ... }})
是total s ($sum: "$total")的$sum
对指定范围的文件(window)
- 在我们的案例中是任何以前的文档:window: { documents: [ "unbounded", "current" ] } }
- 由[ "unbounded", "current" ]定义，意味着窗口是第一个文档(unbounded)和当前文档(current)之间看到的所有文档。

这是另一种方法

管道

db.col.aggregate([
    {$group : {
        _id : { time :{ $dateToString: {format: "%Y-%m-%d", date: "$time", timezone: "-05:00"}}},
        value : {$sum : "$value"}
    }},
    {$addFields : {_id : "$_id.time"}},
    {$sort : {_id : 1}},
    {$group : {_id : null, data : {$push : "$$ROOT"}}},
    {$addFields : {data : {
        $reduce : {
            input : "$data",
            initialValue : {total : 0, d : []},
            in : {
                total : {$sum : ["$$this.value", "$$value.total"]},                
                d : {$concatArrays : [
                        "$$value.d",
                        [{
                            _id : "$$this._id",
                            value : "$$this.value",
                            runningTotal : {$sum : ["$$value.total", "$$this.value"]}
                        }]
                ]}
            }
        }
    }}},
    {$unwind : "$data.d"},
    {$replaceRoot : {newRoot : "$data.d"}}
]).pretty()

集合

> db.col.find()
{ "_id" : ObjectId("4f442120eb03305789000000"), "time" : ISODate("2013-10-10T20:55:36Z"), "value" : 1 }
{ "_id" : ObjectId("4f442120eb03305789000001"), "time" : ISODate("2013-10-11T04:43:16Z"), "value" : 2 }
{ "_id" : ObjectId("4f442120eb03305789000002"), "time" : ISODate("2013-10-12T03:13:06Z"), "value" : 3 }
{ "_id" : ObjectId("4f442120eb03305789000003"), "time" : ISODate("2013-10-11T10:15:38Z"), "value" : 4 }
{ "_id" : ObjectId("4f442120eb03305789000004"), "time" : ISODate("2013-10-13T02:15:38Z"), "value" : 5 }

结果

{ "_id" : "2013-10-10", "value" : 3, "runningTotal" : 3 }
{ "_id" : "2013-10-11", "value" : 7, "runningTotal" : 10 }
{ "_id" : "2013-10-12", "value" : 5, "runningTotal" : 15 }
>

这是一种无需将以前的文档放入新数组然后再处理它们的解决方案。(如果数组太大，那么您可以超过最大BSON文档大小限制，即16MB。)

计算运行总数就像这样简单:

db.collection1.aggregate(
[
  {
    $lookup: {
      from: 'collection1',
      let: { date_to: '$time' },
      pipeline: [
        {
          $match: {
            $expr: {
              $lt: [ '$time', '$$date_to' ]
            }
          }
        },
        {
          $group: {
            _id: null,
            summary: {
              $sum: '$value'
            }
          }
        }
      ],
      as: 'sum_prev_days'
    }
  },
  {
    $addFields: {
      sum_prev_days: {
        $arrayElemAt: [ '$sum_prev_days', 0 ]
      }
    }
  },
  {
    $addFields: {
      running_total: {
        $sum: [ '$value', '$sum_prev_days.summary' ]
      }
    }
  },
  {
    $project: { sum_prev_days: 0 }
  }
]
)

我们所做的:在查找中，我们选择所有具有较小日期时间的文档，并立即计算总和(使用$group作为查找管道的第二步)。$lookup将值放入数组的第一个元素。取出第一个数组元素，然后计算和:当前值+先前值的和。

如果您想将事务分组为天，然后计算运行总数，那么我们需要将$group插入到开始，并将其插入到$lookup的管道中。

db.collection1.aggregate(
[
  {
    $group: {
      _id: {
        $substrBytes: ['$time', 0, 10]
      },
      value: {
        $sum: '$value'
      }
    }
  },
  {
    $lookup: {
      from: 'collection1',
      let: { date_to: '$_id' },
      pipeline: [
        {
          $group: {
            _id: {
              $substrBytes: ['$time', 0, 10]
            },
            value: {
              $sum: '$value'
            }
          }
        },
        {
          $match: {
            $expr: {
              $lt: [ '$_id', '$$date_to' ]
            }
          }
        },
        {
          $group: {
            _id: null,
            summary: {
              $sum: '$value'
            }
          }
        }
      ],
      as: 'sum_prev_days'
    }
  },
  {
    $addFields: {
      sum_prev_days: {
        $arrayElemAt: [ '$sum_prev_days', 0 ]
      }
    }
  },
  {
    $addFields: {
      running_total: {
        $sum: [ '$value', '$sum_prev_days.summary' ]
      }
    }
  },
  {
    $project: { sum_prev_days: 0 }
  }
]
)

结果是:

{ "_id" : "2013-10-10", "value" : 3, "running_total" : 3 }
{ "_id" : "2013-10-11", "value" : 7, "running_total" : 10 }
{ "_id" : "2013-10-12", "value" : 5, "running_total" : 15 }

相关内容

最新更新

热门标签：