mongodb - 来自具有 MongoDB 聚合的事件集合的线性漏斗，这可能吗？

Question

我有很多事件文档，每个事件都有很多字段，但与我的查询相关的是:

• person_id - 对触发事件的人的引用
• event - 用于标识事件的字符串键
• occurred_at - 事件发生时间的协调世界时

我想实现的是:

• 获取事件键列表，例如`['event_1','event_2', 'event_3']
• 按顺序计算执行每个事件和该事件之前的所有事件的人数，即:
  • 执行 event_1 的人数
  • 执行 event_1 和 event_2 的人数
  • 执行 event_1、event_2、event_3 的人数
  • 等等
• 次要目标是能够获得每个事件的平均 occurred_at 日期，以便我可以计算每个事件之间的平均时间

我得到的最好的是以下两个 map reduce:

db.events.mapReduce(function () {
  emit(this.person_id, {
    e: [{
      e: this.event,
      o: this.occurred_at
    }]
  })
}, function (key, values) {
  return {
    e: [].concat.apply([], values.map(function (x) {
      return x.e
    }))
  }
}, {
  query: {
    account_id: ObjectId('52011239b1b9229f92000003'),
    event: {
      $in: ['event_a', 'event_b', 'event_c','event_d','event_e','event_f']
    }
  },
  out: 'people_funnel_chains',
  sort: { person_id: 1, occurred_at: 1 }
})

然后:

db.people_funnel_chains.mapReduce(function() {
  funnel = ['event_a', 'event_b', 'event_c','event_d','event_e','event_f']
  events = this.value.e;
  for (var e in funnel) {
    e = funnel[e];
    if ((i = events.map(function (x) {
      return x.e
    }).indexOf(e)) > -1) {
      emit(e, { c: 1, o: events[i].o })
      events = events.slice(i + 1, events.length);
    } else {
      break;
    }
  }
}, function(key,values) {
    return {
        c: Array.sum(values.map(function(x) { return x.c })),
        o: new Date(Array.sum(values.map(function(x) { return x.o.getTime() }))/values.length)
    };
}, { out: {inline: 1} })

我想使用聚合框架实时实现这一点，但看不出有什么办法。对于 10 万条记录，这需要 10 秒，我可以增量运行它，这意味着它足够快，可以接收新数据，但如果我想修改原始查询(例如更改事件链)，则无法完成在一个请求中，我希望它能够做到。

使用 Cursor.forEach() 更新

使用 Cursor.forEach() 我已经设法在这方面取得了巨大的改进(基本上消除了对第一个 map reduce 的要求)。

var time = new Date().getTime(), funnel_event_keys = ['event_a', 'event_b', 'event_c','event_d','event_e','event_f'], looking_for_i = 0, looking_for = funnel_event_keys[0], funnel = {}, last_person_id = null;
for (var i in funnel_event_keys) { funnel[funnel_event_keys[i]] = [0,null] };
db.events.find({
  account_id: ObjectId('52011239b1b9229f92000003'),
  event: {
    $in: funnel_event_keys
  }
}, { person_id: 1, event: 1, occurred_at: 1 }).sort({ person_id: 1, occurred_at: 1 }).forEach(function(e) {

  var current_person_id = e['person_id'].str; 

  if (last_person_id != current_person_id) {
    looking_for_i = 0;
    looking_for = funnel_event_keys[0]
  }

  if (e['event'] == looking_for) {
    var funnel_event = funnel[looking_for]
    funnel_event[0] = funnel_event[0] + 1;
    funnel_event[1] = ((funnel_event[1] || e['occurred_at'].getTime()) + e['occurred_at'].getTime())/2;
    looking_for_i = looking_for_i + 1;
    looking_for = funnel_event_keys[looking_for_i]
  }

  last_person_id = current_person_id;
})
funnel;
new Date().getTime() - time;

我想知道内存中数据的自定义是否能够改进这一点？从 MongoDB 中获取成百上千条记录到内存中(在另一台机器上)将成为一个瓶颈，有没有我不知道的技术可以做到这一点？

Answer 1

我写了一个complete answer on my MongoDB blog但总而言之，您需要做的是根据您关心的操作来计划您的操作，将操作字段的值映射到适当的键名中，按人分组汇总他们执行的三个操作(以及可选的次数) ) 然后转换新字段，检查 action2 是否在 action1 之后完成，action3 是否在 action2 之后完成...最后一个阶段只是总结了只做了 1、1、2、1、2 和 1 的人数然后 3.

使用函数生成聚合管道，可以根据传入的操作数组生成结果。

在我的测试用例中，整个管道在 200 毫秒内运行了 40,000 个文档的集合(这是在我的小型笔记本电脑上)。

正如正确指出的那样，我描述的一般解决方案假设虽然一个 Actor 可以多次执行任何 Action ，但他们只能从 Action 1 前进到 Action 2，但他们不能直接从 Action 1 跳到 Action 3(将 Action 顺序解释为描述在完成 action2 之前不能执行 action3 的先决条件)。

事实证明，聚合框架甚至可以用于顺序完全任意的事件序列，但您仍然想知道在某个时刻有多少人执行了序列 action1、action2、action3。

对原始答案所做的主要调整是在中间添加一个额外的两阶段步骤。此步骤展开按人员收集的文档以重新分组，找到第一个操作的第一次出现之后的第二个操作的第一次出现。

一旦我们有了最终比较成为 action1，然后是最早出现的 action2，并将其与最近出现的 action3 进行比较。

它可能可以被概括为处理任意数量的事件，但每超过两个额外的事件都会为聚合增加两个阶段。

这里是 my write-up of the modification of the pipeline以获得您正在寻找的答案。