使用聚合框架或 mapreduce 在 MongoDB 中嵌入事件数据中的文档答案

【问题标题】：Embed documents from event data in MongoDB with aggregation framwork or map/reduce使用聚合框架或 mapreduce 在 MongoDB 中嵌入事件数据中的文档
【发布时间】：2014-08-13 18:55:03
【问题描述】：

我有以下数据结构，并希望将包含 loc 字段的文档嵌入到包含持续时间的文档中，但前提是时间戳在时间戳 (ts) 减去父文档的持续时间（以秒为单位）之内。这可以通过聚合框架实现还是通过 map reduce 实现？

{
    "_id" : ObjectId("53df2a44e6583c76253c9869"),
    "deviceId" : NumberLong(1377700226807),
    "ts" : ISODate("2014-08-04T08:37:55.000Z"),
    "duration" : NumberLong(1642),
}
{
    "_id" : ObjectId("53df2a41e6583c4e243c9869"),
    "deviceId" : NumberLong(1377700226807),
    "ts" : ISODate("2014-08-04T08:37:53.000Z"),
    "loc" : {
        "lon" : 5.1101453,
        "lat" : 52.0625047
    }
}
{
    "_id" : ObjectId("53df2a3fe6583c38203c986a"),
    "deviceId" : NumberLong(1377700226807),
    "ts" : ISODate("2014-08-04T08:37:50.000Z"),
        "loc" : {
            "lon" : 5.1101297,
            "lat" : 52.0625031
        }
}
{
    "_id" : ObjectId("53df2a44e6583c76253c9869"),
    "deviceId" : NumberLong(1377700226807),
    "ts" : ISODate("2014-08-04T06:37:55.000Z"),
    "duration" : NumberLong(3600),
}
{
    "_id" : ObjectId("53df2a38e6583c03253c9869"),
    "deviceId" : NumberLong(1377700226807),
    "ts" : ISODate("2014-08-04T06:37:44.000Z"),
        "loc" : {
            "lon" : 5.1101176,
            "lat" : 52.0625171
        }
}
{
    "_id" : ObjectId("53df2a33e6583c51243c9869"),
    "deviceId" : NumberLong(1377700226807),
    "ts" : ISODate("2014-08-04T06:37:38.000Z"),
        "loc" : {
            "lon" : 5.1101409,
            "lat" : 52.0625818
        }
}
{
    "_id" : ObjectId("53df2a2de6583c38203c9869"),
    "deviceId" : NumberLong(1377700226807),
    "ts" : ISODate("2014-08-04T06:37:32.000Z"),
        "loc" : {
            "lon" : 5.1099513,
            "lat" : 52.0624157
        }
}

这是所需的格式

{
    "_id" : ObjectId("53df2a44e6583c76253c9869"),
    "deviceId" : NumberLong(1377700226807),
    "ts" : ISODate("2014-08-04T08:37:55.000Z"),
    "duration" : NumberLong(1642),
    "data" : [
       {
          "ts" : ISODate("2014-08-04T08:37:53.000Z"),
          "loc" : {
             "lon" : 5.1101453,
             "lat" : 52.0625047
          }
       },
       {
          "ts" : ISODate("2014-08-04T08:37:50.000Z"),
          "loc" : {
            "lon" : 5.1101297,
            "lat" : 52.0625031
           }
       }
    ]
}

【问题讨论】：

你能提供一份文件来说明你想要的输出吗？您要更改集合中的文档还是创建新文档？您是否在一个集合中混合了不同类型的文档？您如何使用这些文件？从表面上看，这似乎是一个最好通过模式设计而不是聚合或 map/reduce 解决的问题。
我添加了示例，我更喜欢修改当前数据，但创建新集合也是一种选择。该集合包含汽车事件数据，事件可以是包含持续时间和结束时间的行程结束事件。它还将包含在骑行过程中发生的位置事件。这些事件存储在同一个集合中。目前我正在通过 cron 处理它，但想知道我是否可以通过聚合来处理它。但是认为这是不可能的，因为没有什么可以做的。

标签： javascript mongodb mapreduce mongodb-query aggregation-framework

【解决方案1】：

使用聚合框架不容易做到这一点，但使用 MapReduce 可以做到。

假设数据是收集的财产（即，没有丢失的“骑行”文档与具有“loc”值的文档的持续时间），您可以这样做：

map=function () {
   var startTime;
   if (this.hasOwnProperty("duration"))
       startTime=this.ts-this.duration*1000;
   else
       startTime=this.ts;
   emit(this.deviceId, {startTs:new Date(startTime), endTs:this.ts, loc:this.loc, duration:this.duration});
}

Map 以标准化格式输出内容，reduce 将它们全部分组到每个 deviceId 的单个数组中。

reduce=function (key,values) {
   var result = { vals : [ ] };
   values.forEach(function(v) {
       result.vals.push(v);
   })
   return result;
}

所有实际处理（对每个 deviceId 进行分组）都发生在 finalize 函数中，该函数为每个 deviceId 获取一个数组，对其进行排序，并分组到您期望的文档中。

finalize=function (key, value) {
    var lastI=-1;
    var result = {rides: [ ] };
    var ride = { };
    value.vals.sort(function(a,b) { return a.startTs.getTime() - b.startTs.getTime(); } );

    for (i=0; i<value.vals.length; i++) {
        if (value.vals[i].loc == null ) {
           if (ride.hasOwnProperty("locations")) {
               result.rides.push(ride);
               ride={};
           }
           ride["start"]=value.vals[i].startTs;
           ride["end"]=value.vals[i].endTs;
           ride["duration"]=value.vals[i].duration;
           ride["locations"]=[];
           lastI=i;
        } else {
           ride.locations.push({ loc: value.vals[i].loc, ts: value.vals[i].endTs});
        }
    }
    result.rides.push(ride);
    return result;
}

我在您的测试数据中添加了几个 deviceId：

db.rides.find({},{_id:0})
{ "deviceId" : NumberLong("1377700226807"), "ts" : ISODate("2014-08-04T06:37:32Z"), "loc" : { "lon" : 5.1099513, "lat" : 52.0624157 } }
{ "deviceId" : NumberLong("1377700226910"), "ts" : ISODate("2014-08-04T06:37:32Z"), "loc" : { "lon" : 5.1099513, "lat" : 52.0624157 } }
{ "deviceId" : NumberLong("1377700226807"), "ts" : ISODate("2014-08-04T06:37:38Z"), "loc" : { "lon" : 5.1101409, "lat" : 52.0625818 } }
{ "deviceId" : NumberLong("1377700226910"), "ts" : ISODate("2014-08-04T06:37:38Z"), "loc" : { "lon" : 5.1101409, "lat" : 52.0625818 } }
{ "deviceId" : NumberLong("1377700226807"), "ts" : ISODate("2014-08-04T06:37:44Z"), "loc" : { "lon" : 5.1101176, "lat" : 52.0625171 } }
{ "deviceId" : NumberLong("1377700226910"), "ts" : ISODate("2014-08-04T06:37:44Z"), "loc" : { "lon" : 5.1101176, "lat" : 52.0625171 } }
{ "deviceId" : NumberLong("1377700226807"), "ts" : ISODate("2014-08-04T06:37:55Z"), "duration" : NumberLong(3600) }
{ "deviceId" : NumberLong("1377700226910"), "ts" : ISODate("2014-08-04T06:37:55Z"), "duration" : NumberLong(3600) }
{ "deviceId" : NumberLong("1377700226807"), "ts" : ISODate("2014-08-04T08:37:50Z"), "loc" : { "lon" : 5.1101297, "lat" : 52.0625031 } }
{ "deviceId" : NumberLong("1377700226908"), "ts" : ISODate("2014-08-04T08:37:50Z"), "loc" : { "lon" : 5.1101297, "lat" : 52.0625031 } }
{ "deviceId" : NumberLong("1377700226807"), "ts" : ISODate("2014-08-04T08:37:53Z"), "loc" : { "lon" : 5.1101453, "lat" : 52.0625047 } }
{ "deviceId" : NumberLong("1377700226908"), "ts" : ISODate("2014-08-04T08:37:53Z"), "loc" : { "lon" : 5.1101453, "lat" : 52.0625047 } }
{ "deviceId" : NumberLong("1377700226807"), "ts" : ISODate("2014-08-04T08:37:55Z"), "duration" : NumberLong(1642) }
{ "deviceId" : NumberLong("1377700226908"), "ts" : ISODate("2014-08-04T08:37:55Z"), "duration" : NumberLong(1642) }

并通过 MR 运行它

db.rides.mapReduce(map, reduce, {out:"newrides", finalize:finalize})
{
        "result" : "frides",
        "timeMillis" : 47,
        "counts" : {
                "input" : 14,
                "emit" : 14,
                "reduce" : 3,
                "output" : 3
        },
        "ok" : 1
}

结果是：

db.newrides.find().pretty()
{
        "_id" : NumberLong("1377700226807"),
        "value" : {
                "rides" : [
                        {
                                "start" : ISODate("2014-08-04T05:37:55Z"),
                                "end" : ISODate("2014-08-04T06:37:55Z"),
                                "duration" : NumberLong(3600),
                                "locations" : [
                                        {
                                                "loc" : {
                                                        "lon" : 5.1099513,
                                                        "lat" : 52.0624157
                                                },
                                                "ts" : ISODate("2014-08-04T06:37:32Z")
                                        },
                                        {
                                                "loc" : {
                                                        "lon" : 5.1101409,
                                                        "lat" : 52.0625818
                                                },
                                                "ts" : ISODate("2014-08-04T06:37:38Z")
                                        },
                                        {
                                                "loc" : {
                                                        "lon" : 5.1101176,
                                                        "lat" : 52.0625171
                                                },
                                                "ts" : ISODate("2014-08-04T06:37:44Z")
                                        }
                                ]
                        },
                        {
                                "start" : ISODate("2014-08-04T08:10:33Z"),
                                "end" : ISODate("2014-08-04T08:37:55Z"),
                                "duration" : NumberLong(1642),
                                "locations" : [
                                        {
                                                "loc" : {
                                                        "lon" : 5.1101297,
                                                        "lat" : 52.0625031
                                                },
                                                "ts" : ISODate("2014-08-04T08:37:50Z")
                                        },
                                        {
                                                "loc" : {
                                                        "lon" : 5.1101453,
                                                        "lat" : 52.0625047
                                                },
                                                "ts" : ISODate("2014-08-04T08:37:53Z")
                                        }
                                ]
                        }
                ]
        }
}
{
        "_id" : NumberLong("1377700226908"),
        "value" : {
                "rides" : [
                        {
                                "start" : ISODate("2014-08-04T08:10:33Z"),
                                "end" : ISODate("2014-08-04T08:37:55Z"),
                                "duration" : NumberLong(1642),
                                "locations" : [
                                        {
                                                "loc" : {
                                                        "lon" : 5.1101297,
                                                        "lat" : 52.0625031
                                                },
                                                "ts" : ISODate("2014-08-04T08:37:50Z")
                                        },
                                        {
                                                "loc" : {
                                                        "lon" : 5.1101453,
                                                        "lat" : 52.0625047
                                                },
                                                "ts" : ISODate("2014-08-04T08:37:53Z")
                                        }
                                ]
                        }
                ]
        }
}
{
        "_id" : NumberLong("1377700226910"),
        "value" : {
                "rides" : [
                        {
                                "start" : ISODate("2014-08-04T05:37:55Z"),
                                "end" : ISODate("2014-08-04T06:37:55Z"),
                                "duration" : NumberLong(3600),
                                "locations" : [
                                        {
                                                "loc" : {
                                                        "lon" : 5.1099513,
                                                        "lat" : 52.0624157
                                                },
                                                "ts" : ISODate("2014-08-04T06:37:32Z")
                                        },
                                        {
                                                "loc" : {
                                                        "lon" : 5.1101409,
                                                        "lat" : 52.0625818
                                                },
                                                "ts" : ISODate("2014-08-04T06:37:38Z")
                                        },
                                        {
                                                "loc" : {
                                                        "lon" : 5.1101176,
                                                        "lat" : 52.0625171
                                                },
                                                "ts" : ISODate("2014-08-04T06:37:44Z")
                                        }
                                ]
                        }
                ]
        }
}

【讨论】：

非常感谢，您和 Neil 的回答让我对如何使用 map reduce 有了更好的理解，我自己已经尝试过了，但还没有走到这一步。我的数据集可能缺少骑行事件（它们稍后会出现），因此会有包含没有相应骑行的 loc 的文档。我想我将能够根据您的回答自己解决这个问题。谢谢！

【解决方案2】：

在处理管道时，聚合框架无法跨文档“保留”信息。因此，您所描述的那种“父/子”关系是通过与未在文档中直接指定的“父”文档进行比较来决定的。

然而，mapReduce 方法可以访问“全局”范围的变量。这可以允许“识别”父级详细信息，然后可以将其存储在变量中，以便根据需要与可能的子级进行比较。

db.collection.mapReduce(
  function() {
    lastValue.data = [];

    if ( lastId == null || this.hasOwnProperty("duration") ) {
      lastId = this._id;
      lastValue.deviceId = this.deviceId;
      lastValue.ts = this.ts;
      lastValue.duration = this.duration;
    }

    if (
        ( this.hasOwnProperty("loc") ) &&
        (
          ( lastValue.ts.valueOf() - ( lastValue.duration * 1000 ) ) <
          this.ts.valueOf()
        )
      )
      lastValue.data.push({
        "ts": this.ts,
        "loc": this.loc
      });


    emit ( lastId, lastValue );

  },
  function (key,values) {

    var reduced = {};

    values.forEach(function(value) {
      if ( !reduced.hasOwnProperty("deviceId") ) {
        reduced.deviceId = value.deviceId;
        reduced.ts = value.ts;
        reduced.duration = value.duration;
        reduced.data = [];
      }

      value.data.forEach(function(dat) {
        reduced.data.push(dat);
      });

    });

    return reduced;

  },
  { 
      "sort": { "ts": -1 },
      "scope": { "lastId": null, "lastValue": {} }, 
      "out": { "inline": 1 }
  }
)

所以本质上这是使用 mapReduce 可用的“范围”声明“存储”要作为“lastId”发出的“键”。清除“父”文档包含一个持续时间，所以这里有一些东西可以使用。

“数据”元素需要作为数组发出。这是由于文档中描述的requirement of the "reduce" function。关键是输入必须与预期输入的格式相同。每个“key”可以多次调用“reduce”函数。

“mapper”中的条件可以在调用“reduce”函数之前限制此数组中存在的值。这也避免了在只发出一个“key”的情况下依赖添加“finalize”方法，并且“reduce”功能不会触及该项目。因此，预先进行过滤的工作量更少。

“reducer”现在被降级为仅“合并”结果以组合“data”中符合“children”条件的元素。

自然地，排序顺序是“降序”时间戳或“ts”值。这将是一个很好的索引点，并且顺序与处理文档进行比较的方式保持一致，因此每个“断点”在检测到新父级时都是有效的。

输出当然是 mapReduce 样式。因此，要获得“确切”所需的输出将需要更多的后期处理，但正如您所见，这基本上是结果：

    "results" : [
        {
            "_id" : ObjectId("53f15b8beb75bdce84f914a1"),
            "value" : {
                "deviceId" : NumberLong("1377700226807"),
                "ts" : ISODate("2014-08-04T08:37:55Z"),
                "duration" : NumberLong(1642),
                "data" : [
                    {
                        "ts" : ISODate("2014-08-04T08:37:53Z"),
                        "loc" : {
                            "lon" : 5.1101453,
                            "lat" : 52.0625047
                        }
                    },
                    {
                        "ts" : ISODate("2014-08-04T08:37:50Z"),
                        "loc" : {
                            "lon" : 5.1101297,
                            "lat" : 52.0625031
                        }
                    }
                ]
             }
        },
        {
            "_id" : ObjectId("53f15b8beb75bdce84f914a4"),
            "value" : {
                "deviceId" : NumberLong("1377700226807"),
                "ts" : ISODate("2014-08-04T06:37:55Z"),
                "duration" : NumberLong(3600),
                "data" : [
                    {
                        "ts" : ISODate("2014-08-04T06:37:44Z"),
                        "loc" : {
                            "lon" : 5.1101176,
                            "lat" : 52.0625171
                        }
                    },
                    {
                        "ts" : ISODate("2014-08-04T06:37:38Z"),
                        "loc" : {
                            "lon" : 5.1101409,
                            "lat" : 52.0625818

                    },
                    {
                        "ts" : ISODate("2014-08-04T06:37:32Z"),
                        "loc" : {
                            "lon" : 5.1099513,
                            "lat" : 52.0624157
                        }
                    }
                ]
            }
        }
    ]

请注意，此处给出的输出 _id 值与示例不同，因为示例包含 _id 的“重复”值，这当然是不允许在主键上的。

【讨论】：