我们可以使用MongoDB 3.2或更高版本中的聚合框架来实现,因为我们真正需要的是一种跟踪数组中当前和上一个元素的索引的方法,幸运的是从MongoDB 3.2开始,我们可以使用

$unwind运算符来解构我们的数组,并通过指定一个文档作为操作数,而不是以$为前缀的传统“路径”,包括数组中每个元素的索引.

从那里我们有两个选择.第一个是MongoDB 3.2,第二个是即将发布的MongoDB(在撰写本文时).

下一步在管道中,我们需要$group我们的文档,并使用$push累加器运算符返回一个如下所示的子文档数组：

{

"_id" : ObjectId("57c11ddbe860bd0b5df6bc64"),

"time_series" : [

{ "value" : 10, "index" : NumberLong(0) },

{ "value" : 20, "index" : NumberLong(1) },

{ "value" : 40, "index" : NumberLong(2) },

{ "value" : 70, "index" : NumberLong(3) },

{ "value" : 110, "index" : NumberLong(4) }

]

}

终于到了$project阶段.在这个阶段,我们需要使用$map运算符将一系列表达式应用于$group阶段新计算的数组中的每个元素.

这是在$map里面发生了什么(请参阅$map作为for循环)在表达式中：

对于每个子文档,我们使用$let变量运算符将值字段分配给变量.然后,我们从数组中下一个元素的“值”字段的值中减去它的值.

由于数组中的下一个元素是当前索引的元素加1,所以我们需要的是$arrayElemAt运算符的帮助和当前元素索引的简单$add和1.

$subtract表达式返回一个负值,因此我们需要使用$multiply运算符将值乘以-1.

我们还需要$filter所得到的数组,因为最后一个元素是None或null.原因是当当前元素是最后一个元素时,$subtract返回None,因为下一个元素的索引等于数组的大小.

db.collection.aggregate(

[

{ "$unwind": {

"path": "$time_series",

"includeArrayIndex": "index"

}},

{ "$group": {

"_id": "$_id",

"time_series": {

"$push": {

"value": "$time_series",

"index": "$index"

}

}

}},

{ "$project": {

"time_series": {

"$filter": {

"input": {

"$map": {

"input": "$time_series",

"as": "el",

"in": {

"$multiply": [

{ "$subtract": [

"$$el.value",

{ "$let": {

"vars": {

"nextElement": {

"$arrayElemAt": [

"$time_series",

{ "$add": [

"$$el.index",

1

]}

]}

},

"in": "$$nextElement.value"

}

}

]},

-1

]

}

}

},

"as": "item",

"cond": { "$gte": [ "$$item", 0 ] }

}

}

}}

]

)

在即将推出的版本中将提供另一种选择.

首先在$group阶段,我们返回两个不同的数组.一个用于元素,另一个用于索引,然后$zip如下所示的两个数组：here.从他们,我们只需使用整数索引访问每个元素,而不是将其值分配给带有$let的变量.

db.collection.aggregate(

[

{ "$unwind": {

"path": "$time_series",

"includeArrayIndex": "index"

}},

{ "$group": {

"_id": "$_id",

"values": { "$push": "$time_series" },

"indexes": { "$push": "$index" }

}},

{ "$project": {

"time_series": {

"$filter": {

"input": {

"$map": {

"input": {

"$zip": {

"inputs": [

"$values",

"$indexes"

]

}

},

"as": "el",

"in": {

"$multiply": [

{ "$subtract": [

{ "$arrayElemAt": [

"$$el",

0

]},

{ "$arrayElemAt": [

"$values",

{ "$add": [

{ "$arrayElemAt": [

"$$el",

1

]},

1

]}

]}

]},

-1

]

}

}

},

"as": "item",

"cond": { "$gte": [ "$$item", 0 ] }

}

}

}}

]

)

请注意,我们也可以在$project stage的早期使用$reverse来反转阵列,如图here所示,以避免使用$multiply.

这两个查询都产生如下：

{

"_id" : ObjectId("57c11ddbe860bd0b5df6bc64"),

"time_series" : [ 10, 20, 30, 40 ]

}

我认为效率较低的另一个选项是使用map_reduce方法对我们的收集执行地图/缩小操作.

>>> import pymongo

>>> from bson.code import Code

>>> client = pymongo.MongoClient()

>>> db = client.test

>>> collection = db.collection

>>> mapper = Code("""

... function() {

... var derivatives = [];

... for (var index=1; index

... derivatives.push(this.time_series[index] - this.time_series[index-1]);

... }

... emit(this._id, derivatives);

... }

... """)

>>> reducer = Code("""

... function(key, value) {}

... """)

>>> for res in collection.map_reduce(mapper, reducer, out={'inline': 1})['results']:

... print(res) # or do something with the document.

...

{'value': [10.0, 20.0, 30.0, 40.0], '_id': ObjectId('57c11ddbe860bd0b5df6bc64')}

您还可以检索所有文档,并使用numpy.diff返回导数,如下所示：

import numpy as np

for document in collection.find({}, {'time_series': 1}):

result = np.diff(document['time_series'])

现在怎么样一点基准：

机：

OS: Ubuntu 16.04

Memory: 15.6 GiB

Processor: Intel® Xeon(R) CPU E3-1231 v3 @ 3.40GHz × 8

在我的机器上按照这个顺序运行的三个查询分别给出以下结果：

基准测试结果与500个文件：

MongoDB 3.2

100 loops, best of 3: 2.32 ms per loop

MongoDB 3.3.11

1000 loops, best of 3: 1.72 ms per loop

MapReduce的

100 loops, best of 3: 15.7 ms per loop

100 loops, best of 3: 3.61 ms per loop

结论

即使解决方案不明显,使用聚合也是预期的最佳选择.

由于JavaScript评估,mapReduce解决方案是微不足道的,但效率很低.

*您可以通过安装MongoDB的当前开发版本(作为本写作时间)来测试第二个查询.