如何将Mongodb时间点数据分组到连续的时间组？

Question

我有一个类似于以下形式的决策文件集合:

    { 
    _id: ObjectId("23de23802fe925b6ef7162a4"),
    userId: 6de4, 
    decision: true, 
    datetime:ISODate("2016-07-27T08:22:47.169Z")
    },
    { 
    _id: ObjectId("507f1f77bcf86cd799439011"),
    userId: 23f4, 
    decision: true, 
    datetime:ISODate("2016-02-03T11:48:50.456Z")
    },
.
.
.

我正在努力想出一种将这些文档分组为连续日期时间组的方法.即文档应该属于特定组,如果它小于,比如组中至少一个其他文档的5分钟.

目标是实现在"会话"中做出的决策群.然后可以使用聚合(例如每个决策的平均时间等)对这些"会话"组进行进一步的见解.

如果不能使用MongoDb的聚合框架,可以使用map-reduce或其他方法完成.我愿意接受建议.

澄清

描述问题的另一种方法是将以下算法应用于文档集合.

1. 首先将文档放入日期时间顺序.
2. 将最早出现的文件(按时间顺序排列)放在自己的 组中,然后转到下一个文档.
3. 如果下一个文档的日期时间小于前一个文档之后的指定时间(比如5分钟),则将其放在与之前相同的组中.如果没有,请创建一个新组并将此文档放入其中.
4. 重复步骤3.直到遍历所有文档.

这将使集合具有所需的"会话"分组.当然,这只是描绘问题的一种方式.我不知道有什么方法可以遍历有序集合,同时使用MongoDb以这种方式进行分组.

可以这样做吗？有没有其他方法可以使用MongoDb获得相同的结果？

Answer 1

根据您描述的算法，每个文档的分组逻辑始终依赖于另一个文档。我没有找到使用映射缩减、聚合或单个 MongoDB 查询来执行此操作的方法。我看到的唯一解决方案是严格遵循您的算法，即阅读每个文档并决定它是否属于当前组或是否应该属于新组。

不建议将所有文档加载到内存中，因为它可能是一个非常大的集合。所以我使用流来逐个文档加载文档。

创建一个游标来查找所有文档并按日期对它们进行排序，然后用于cursor.on('data', function(document){ ... });单独读取每个文档。

var groups = {} // init group object
var currentGroupKey;
var groupInterval = 5 * 60 * 1000; // Five minutes in milliseconds

var cursor = db.collection("documents").find({}).sort({date: 1});

cursor.on('data', function(doc) {
  var timestamp = doc.date.getTime();

  if (currentGroupKey != null && currentGroupKey + groupInterval >= timestamp) {
    // add it to current group
    groups[currentGroupKey].push(doc);
  } else {
    // create a new group
    groups[timestamp] = [doc];
    currentGroupKey = timestamp;
  }
});
cursor.once('end', function() {
  // This is called after last document is read
  console.log(groups); // print your grouped documents
  db.close();
});

对于本文件

[ { _id: 57f59acb8e73d9634ac8c7b0,
    index: 3,
    date: Wed Oct 05 2016 21:02:29 GMT-0300 (BRT) },
  { _id: 57f59acb8e73d9634ac8c7ae,
    index: 1,
    date: Wed Oct 05 2016 21:04:02 GMT-0300 (BRT) },
  { _id: 57f59acb8e73d9634ac8c7b3,
    index: 6,
    date: Wed Oct 05 2016 21:07:43 GMT-0300 (BRT) },
  { _id: 57f59acb8e73d9634ac8c7b4,
    index: 7,
    date: Wed Oct 05 2016 21:10:26 GMT-0300 (BRT) },
  { _id: 57f59acb8e73d9634ac8c7b2,
    index: 5,
    date: Wed Oct 05 2016 21:14:23 GMT-0300 (BRT) },
  { _id: 57f59acb8e73d9634ac8c7b5,
    index: 8,
    date: Wed Oct 05 2016 21:17:39 GMT-0300 (BRT) },
  { _id: 57f59acb8e73d9634ac8c7b6,
    index: 9,
    date: Wed Oct 05 2016 21:21:07 GMT-0300 (BRT) },
  { _id: 57f59acb8e73d9634ac8c7ad,
    index: 0,
    date: Wed Oct 05 2016 21:24:19 GMT-0300 (BRT) },
  { _id: 57f59acb8e73d9634ac8c7af,
    index: 2,
    date: Wed Oct 05 2016 21:25:50 GMT-0300 (BRT) },
  { _id: 57f59acb8e73d9634ac8c7b1,
    index: 4,
    date: Wed Oct 05 2016 21:28:13 GMT-0300 (BRT) } ]

最终的组对象是

{ '1475712149573':
   [ { _id: 57f59acb8e73d9634ac8c7b0,
       index: 3,
       date: Wed Oct 05 2016 21:02:29 GMT-0300 (BRT) },
     { _id: 57f59acb8e73d9634ac8c7ae,
       index: 1,
       date: Wed Oct 05 2016 21:04:02 GMT-0300 (BRT) } ],
  '1475712463238':
   [ { _id: 57f59acb8e73d9634ac8c7b3,
       index: 6,
       date: Wed Oct 05 2016 21:07:43 GMT-0300 (BRT) },
     { _id: 57f59acb8e73d9634ac8c7b4,
       index: 7,
       date: Wed Oct 05 2016 21:10:26 GMT-0300 (BRT) } ],
  '1475712863890':
   [ { _id: 57f59acb8e73d9634ac8c7b2,
       index: 5,
       date: Wed Oct 05 2016 21:14:23 GMT-0300 (BRT) },
     { _id: 57f59acb8e73d9634ac8c7b5,
       index: 8,
       date: Wed Oct 05 2016 21:17:39 GMT-0300 (BRT) } ],
  '1475713267412':
   [ { _id: 57f59acb8e73d9634ac8c7b6,
       index: 9,
       date: Wed Oct 05 2016 21:21:07 GMT-0300 (BRT) },
     { _id: 57f59acb8e73d9634ac8c7ad,
       index: 0,
       date: Wed Oct 05 2016 21:24:19 GMT-0300 (BRT) },
     { _id: 57f59acb8e73d9634ac8c7af,
       index: 2,
       date: Wed Oct 05 2016 21:25:50 GMT-0300 (BRT) } ],
  '1475713693672':
   [ { _id: 57f59acb8e73d9634ac8c7b1,
       index: 4,
       date: Wed Oct 05 2016 21:28:13 GMT-0300 (BRT) } ] }

编辑

由于分组的逻辑始终是最后读取的文档，因此我修改了算法以适应它。现在它还使用组密钥更新每个文档，因此不会将所有文档加载到内存中。

var lastDocumentTimestamp;
var groupIndex = 0;
var groupInterval = 5 * 60 * 1000; // Five minutes in milliseconds

var cursor = db.collection("documents").find({}).sort({date: 1});

cursor.on('data', function(doc) {
  var timestamp = doc.date.getTime();

  if (lastDocumentTimestamp + groupInterval < timestamp) {
    groupIndex++;
  }
  lastDocumentTimestamp = timestamp;
  db.collection("documents").update({ _id: doc._id}, { $set: {group: groupIndex}});
});
cursor.once('end', function() {
  // This is called after last document is read
  db.close();
});

之后，您可以使用聚合按组对文档进行分组

db.collection("documents").aggregate([{
  $group: {
    _id: "$group",
    count: { $sum: 1 },
    docs: { $push: "$date" }
  }
}])

这会产生如下结果：

[ { _id: 0,
    count: 1,
    docs: [ Thu Oct 06 2016 22:00:20 GMT-0300 (BRT) ] },
  { _id: 1,
    count: 4,
    docs:
     [ Thu Oct 06 2016 22:20:31 GMT-0300 (BRT),
       Thu Oct 06 2016 22:22:52 GMT-0300 (BRT),
       Thu Oct 06 2016 22:25:34 GMT-0300 (BRT),
       Thu Oct 06 2016 22:27:15 GMT-0300 (BRT) ] },
  { _id: 2,
    count: 5,
    docs:
     [ Thu Oct 06 2016 22:33:27 GMT-0300 (BRT),
       Thu Oct 06 2016 22:35:45 GMT-0300 (BRT),
       Thu Oct 06 2016 22:38:45 GMT-0300 (BRT),
       Thu Oct 06 2016 22:40:02 GMT-0300 (BRT),
       Thu Oct 06 2016 22:44:20 GMT-0300 (BRT) ] } ]