MongoDB 聚合：从先前行的总和计算运行总计答案

【问题标题】：MongoDB Aggregation: Compute Running Totals from sum of previous rowsMongoDB 聚合：从先前行的总和计算运行总计
【发布时间】：2013-04-17 22:52:25
【问题描述】：

示例文件：

{ 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 聚合可以做到这一点吗？

【问题讨论】：

你能保持一个运行的总数吗？这将是最简单和最有效的，特别是因为数据没有改变。聚合框架是动态计算这类静态数据的一种非常昂贵的方法。
目前没有办法用聚合框架做到这一点。
@cirrus 感谢您的回答。不过我不太确定该怎么做...
嗯，我认为这将涉及几个聚合查询。你不能在一个命令中做到这一点。但是，通过在进行时进行计算，这仅意味着在每个条目中添加一个其他字段以跟踪运行总数。根据您的应用程序，您可以在写入数据时执行此操作，或者您可以在每天结束时运行后台任务来计算它。但这假设您正在编写数据，我不知道您的数据来自哪里。如果数据已经存在，则您必须每天运行一个查询并将其存储在其他地方。

标签： mongodb aggregation-framework mongoid cumulative-sum

【解决方案1】：

从 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这个阶段：

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

【讨论】：

【解决方案2】：

这是一个解决方案，无需将以前的文档推送到新数组中，然后对其进行处理。（如果数组变得太大，那么您可能会超过最大 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 }

【讨论】：

【解决方案3】：

这是另一种方法

管道

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 }
>

【讨论】：

【解决方案4】：

这可以满足您的需求。我已经对数据中的时间进行了标准化，因此它们组合在一起（您可以执行类似this 的操作）。这个想法是$group 并将time 和total 推送到单独的数组中。然后$unwindtime 数组，您为每个time 文档制作了totals 数组的副本。然后，您可以从包含不同时间所有数据的数组中计算runningTotal（或类似滚动平均值）。 $unwind 生成的“索引”是与 time 对应的 total 的数组索引。在$unwinding 之前$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 ] } ] } },
            }
        },
    ]
);

我在一个包含约 80 000 个文档的集合上使用了类似的东西，总计 63 个结果。我不确定它在更大的集合上的效果如何，但我发现一旦数据减少到可管理的大小，对聚合数据执行转换（投影、数组操作）似乎不会产生很大的性能成本。

【讨论】：