聚合时间戳周期的不同状态

我正在使用mongoDB，需要对一段时间内（每小时或每月…）初始化、执行和完成的搜索nº操作进行查询

我的json文档有以下结构：

{
  "_id" : ObjectId("55263d62c63265b9bb138551"),
  "timestamp" : ISODate("2015-02-12T15:27:48.546Z"),       
  "duration" : 199821
}

timestamp字段是procces init start，duration

我可以用这个查询对一段时间内的操作次数（10分钟）进行分组：

db.test.aggregate([ { "$match" :{ "timestamp":{ "$gte": ISODate("2015-0427T12:00:00.0Z") } }}, { "$group" :{ "_id": { "dayOfMonth":{ "$dayOfMonth": "$timestamp" }, "month":{ "$month": "$timestamp" }, "hour": { "$hour":"$timestamp" }, "time": { "$subtract": [ { "$minute":"$timestamp" }, { "$mod": [{ "$minute": "$timestamp" }, 10] } ] } }, "count":{ "$sum":1 } }}, { "$sort": { "_id.time": 1 } } ])
但我也需要数量的"执行中"，和"完成"。
我尝试了mapreduce和其他聚合查询，但无法获得类似于的结果

{ _id: { "month" : 03, "minute" : 00, "Initialized" : 6, "InExecution" : 10, "Finished": 5 } _id: { "month" : 03, "minute" : 10, "Initialized" : 4, "InExecution" : 12, "Finished": 4 } _id: { "month" : 03, "minute" : 20, "Initialized" : 3, "InExecution" : 8, "Finished": 5 } }

这是一个很难理解的问题，但这里有一个主要问题，即"聚合框架"，它主要是您的"活动"根据其当前状态生活在几个不同的时间间隔中。

这意味着它总是有一个"开始"one_answers"完成"的间隔，以及可能的"几个"间隔，在这些间隔中任务可以被视为"正在执行"。

聚合框架不能在一次过程中真正做到这一点。但你可以使用mapReduce:

db.test.mapReduce(
  function() {
     // Work out time values
     var finished = this.timestamp.valueOf() + this.duration,
         finishedInterval = finished -
           ( finished % ( 1000 * 60 * 10 ) ),
         interval = this.timestamp.valueOf() -
           ( this.timestamp.valueOf() % ( 1000 * 60 * 10 ) );
     // Emit initialized
     emit(       
       {
         "year": new Date(interval).getUTCFullYear(),
         "month": new Date(interval).getUTCMonth()+1,
         "day": new Date(interval).getUTCDate(),
         "hour": new Date(interval).getUTCHours(),
         "minute": new Date(interval).getUTCMinutes()
       },
       {
           "Initialized": 1,
           "InExecution": 0,
           "Finshed": 0
       }
     );
     // Emit finished
     emit(       
       {
         "year": new Date(finishedInterval).getUTCFullYear(),
         "month": new Date(finishedInterval).getUTCMonth()+1,
         "day": new Date(finishedInterval).getUTCDate(),
         "hour": new Date(finishedInterval).getUTCHours(),
         "minute": new Date(finsihedInterval).getUTCMinutes()
       },
       {
           "Initialized": 0,
           "InExecution": 0,
           "Finshed": 1
       }
     );
     // Emit In execution for every 10 minute interval until finished
     if ( ( interval + ( 1000 * 60 * 10 ) ) < finishedInterval ) {
       for ( var x = interval; x<finishedInterval; x+= ( 1000 * 60 * 10 ) ) {
         emit(
           {
             "year": new Date(x).getUTCFullYear(),
             "month": new Date(x).getUTCMonth()+1,
             "day": new Date(x).getUTCDate(),
             "hour": new Date(x).getUTCHours(),
             "minute": new Date(x).getUTCMinutes()
           },
           {
             "Initialized": 0,
             "InExecution": 1,
             "Finshed": 0
           }
         );
       }
     }
  },
  function(key,values) {
    var result = { "Initialized": 0, "InExecution": 0, "Finshed": 0 };
    values.forEach(function(value) {
      Object.keys(value).forEach(function(key) {
          result[key] += value[key];          
      });         
    });
    return result;
  },
  { 
    "out": { "inline": 1 },
    "query": { "timestamp": { "$gte": new Date("2015-04-27T12:00:00Z") } }
  }
)

正如您所看到的，大部分工作都是在映射器中完成的。这基本上计算出任务"开始"one_answers"结束"的时间间隔，并为此发出适当的数据。

当然，通过从任务的"开始"间隔开始工作，每10分钟间隔发出一个"执行中"计数，而该值小于任务的"结束"间隔。

reducer只需获取每个间隔的所有发射计数并将其相加。所以这是一个非常简单的操作。

map和reduce逻辑是合理的，但查询选择逻辑存在问题，因为"finshing"或"in execution"作业可能会在第一个查询时间之前启动。

为了做到这一点，您需要修复查询选择以考虑这一点。由于您没有存储"完成"时间，因此您需要计算这一时间，这意味着使用$where:在查询中进行JavaScript评估

{
  "out": { "inline": 1 },
  "query": {
    "$where": function() {
      return (this.timestamp >= new Date("2015-04-27T12:00:00Z") ||
        new Date(this.timestamp.valueOf() + this.duration) >=
          new Date("2015-04-27T12:00:00Z"))
    }
  }
}

它会拾取在查询开始时间之前仍在运行或在该时间结束的项。

这并不好，因为这会扫描集合，所以最好在数据中包含"finshed"作为一个值，以使查询选择更容易：

{
  "out": { "inline": 1 },
  "query": {
     "$or": [
         { "timestamp": { "$gte": new Date("2015-04-27T12:00:00Z") } },
         { "finished": { "$gte": new Date("2015-04-27T12:00:00Z") } }
     ]
  }
}

这可以使用"索引"，而且速度更快。

最后，这里的"时间戳"过滤器值"之前"仍然会有值发出，因为任何一种形式的"完成"都意味着在此时间之前开始的任务。出于同样的原因，在查询条件和逻辑上设置一个"结束"时间也是一个好主意。

为此，再次更改选项块，以包括要在执行逻辑中使用的"范围"变量，并添加到"查询"条件中：

{
  "out": { "inline": 1 },
  "query": {
    "$or": [
      { 
        "timestamp": { 
          "$gte": new Date("2015-04-27T12:00:00Z"),
          "$lt": new Date("2015-04-28T12:00:00Z")
        }
      },
      { 
        "finished": { 
          "$gte": new Date("2015-04-27T12:00:00Z"),
          "$lt": new Date("2015-04-28T12:00:00Z")
        }
      }
    ]
  },
  "scope": {
      "start": new Date("2015-04-27T12:00:00Z"),
      "finsh": new Date("2015-04-28T12:00:00Z")
  }
}

然后在每次发射周围添加条件，首先针对"间隔"大于"开始"的开始：

     // Emit initialized
     if ( interval >= start.valueOf() ) {
       emit(

在"finishedInterval"小于"finish"的情况下：

     // Emit finished
     if ( finishedInterval <= finish.valueOf() ) {
       emit(

然后限制"执行中"的循环：

     // Emit In execution for every 10 minute interval until finished
     if ( ( interval + ( 1000 * 60 * 10 ) ) < finishedInterval ) {
     for ( var x = interval; (( x<finishedInterval ) && ( x<finish.valueOf() )); x+= ( 1000 * 60 * 10 ) ) {
       if ( x > start.valueOf() ) {
         emit(

这为您提供了一个清晰的起点和终点，同时在结果中列出了所有可能的统计数据。

非常感谢Blakes，

为了你的巨大兴趣。我一直在研究你的解决方案，认为这是个好主意。

我找到了一个可能的聚合框架解决方案。

db.getCollection('test').aggregate(
{$match:{ 
           "timestamp":{$exists: true, "$gte": ISODate("2015-03-27T12:00:00.0Z") },              
         } },
{ $project: {
    _id: 1,
    timestamp : 1,
    error: {$cond: [{$eq: ["$severidad", "ERROR"]}, 1, 0]},
    init: {$cond: [ {$and : [{$eq: [{"$subtract": [
                                    {"$minute":"$timestamp"},
                                    {"$mod": [{"$minute":"$timestamp"}, 10]}
                                ]}, {"$subtract": [
                                    {"$minute":"$timestamp"},
                                    {"$mod": [{"$minute":"$timestamp"}, 10]}
                                ]}]}, {$ne: ["$severidad", "ERROR"]}]}, 1, 0]},
    executing: {$cond: [ {$and : [{$gt: [{"$subtract": [
                                    {"$minute":{ $add: [ "$timestamp", "$datos_aplicacion.duracion"]}},
                                    {"$mod": [{"$minute":{ $add: [ "$timestamp", "$datos_aplicacion.duracion"]}}, 10]}
                                ]}, {"$subtract": [
                                    {"$minute":"$timestamp"},
                                    {"$mod": [{"$minute":"$timestamp"}, 10]}
                                ]}]}, {$ne: ["$severidad", "ERROR"]}]}, 1, 0]},
    finished: {$cond: [ {$and : [{$eq: [{"$subtract": [
                                    {"$minute":{ $add: [ "$timestamp", "$datos_aplicacion.duracion"]}},
                                    {"$mod": [{"$minute":{ $add: [ "$timestamp", "$datos_aplicacion.duracion"]}}, 10]}
                                ]}, {"$subtract": [
                                    {"$minute":"$timestamp"},
                                    {"$mod": [{"$minute":"$timestamp"}, 10]}
                             ]}]}, {$ne: ["$severidad", "ERROR"]}]}, 1, 0]},                                    
    }}, 
{$group :{_id: { 
        dayOfMonth:{"$dayOfMonth":"$timestamp"}, month:{"$month":"$timestamp"}, hour:{"$hour":"$timestamp"} ,
        time: {
                "$subtract": [
                    {"$minute":"$timestamp"},
                    {"$mod": [{"$minute":"$timestamp"}, 10]}
                ]
            },            
    },
    NumError: {$sum:"$error"},
    NumInit:{$sum:"$init"},
    NumExecuting:{$sum:"$executing"},
    NumFinished:{$sum:"$finished"}        
    }},
{ $sort : { "_id": 1} });

使用1M的记录需要1,2秒

问候，

相关内容

最新更新

热门标签：