如何将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或其他方式完成。我愿意听取建议。

<标题>澄清h1> 述这个问题的另一种方法是将下面的算法应用于文档集合。

1. 开始按日期时间顺序排列文件。
2. 将最早出现的文件(按时间顺序)放在自己的文件中组并移动到下一个文档
3. 如果下一个文档的日期时间在紧接在之前的一个(例如5分钟)将其放置在与之前相同的组中。如果没有，创建一个新组并将此文档放入其中。
4. 重复步骤3。直到遍历所有文档。

这将使集合保留所需的"session";分组。当然，这只是描述问题的一种方式。我不知道有什么方法可以遍历一个有序的集合，同时使用MongoDb以这种方式分组。

可以这样做吗?是否有另一种方法来实现使用MongoDb相同的结果?

Answer 1

根据您描述的算法，每个文档的分组逻辑总是依赖于另一个文档。我没有看到使用map reduce，聚合或单个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) ] } ]