use*_*666 13 python mapreduce mongodb pymongo aggregation-framework
是否可以使用聚合框架计算一阶导数?
例如,我有数据:
{time_series : [10,20,40,70,110]}
Run Code Online (Sandbox Code Playgroud)
我正在尝试获得如下输出:
{derivative : [10,20,30,40]}
Run Code Online (Sandbox Code Playgroud)
db.collection.aggregate(
[
{
"$addFields": {
"indexes": {
"$range": [
0,
{
"$size": "$time_series"
}
]
},
"reversedSeries": {
"$reverseArray": "$time_series"
}
}
},
{
"$project": {
"derivatives": {
"$reverseArray": {
"$slice": [
{
"$map": {
"input": {
"$zip": {
"inputs": [
"$reversedSeries",
"$indexes"
]
}
},
"in": {
"$subtract": [
{
"$arrayElemAt": [
"$$this",
0
]
},
{
"$arrayElemAt": [
"$reversedSeries",
{
"$add": [
{
"$arrayElemAt": [
"$$this",
1
]
},
1
]
}
]
}
]
}
}
},
{
"$subtract": [
{
"$size": "$time_series"
},
1
]
}
]
}
},
"time_series": 1
}
}
]
)
Run Code Online (Sandbox Code Playgroud)
我们可以在版本3.4+中使用上面的管道来执行此操作.在管道中,我们使用$addFields管道阶段.运算符添加"time_series"的元素索引的数组来做文档,我们还颠倒了时间序列数组并分别使用$range和$reverseArray运算符将它添加到文档中
我们在这里颠倒了数组,因为数组中位置的元素p总是大于位置处的元素,p+1这意味着[p] - [p+1] < 0我们不想使用$multiply这里.(参见版本3.2的管道)
接下来,我们$zipped使用索引数组的时间序列数据,并substract使用$map运算符将表达式应用于结果数组.
然后我们$slice将结果丢弃null/None数组中的值并重新反转结果.
在3.2中,我们可以使用$unwind运算符来展开我们的数组,并通过将文档指定为操作数而不是以$为前缀的传统"路径"来包含数组中每个元素的索引.
接下来在管道中,我们需要$group我们的文档并使用$pushaccumulator操作符返回一个子文档数组,如下所示:
{
"_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) }
]
}
Run Code Online (Sandbox Code Playgroud)
终于到了$project舞台.在这个阶段,我们需要使用$map运算符将一系列表达式应用于$group舞台中新计算的数组中的每个元素.
这里是被里面的事情$map(见$map一个for循环)的表达式:
对于每个子文档,我们使用变量运算符将值字段分配给$let变量.然后,我们从数组中下一个元素的"value"字段的值中减去它的值.
由于数组中的下一个元素是当前索引加上1的元素,我们所需要的只是$arrayElemAt运算符的帮助和$add当前元素索引的简单概念1.
该$subtract表达式返回一个负值,所以我们需要通过乘以值-1使用的$multiply运营商.
我们还需要$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
]
}
}
}
}
}
])
Run Code Online (Sandbox Code Playgroud)
我认为效率较低的另一个选项是使用该map_reduce方法对我们的集合执行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<this.time_series.length; 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')}
Run Code Online (Sandbox Code Playgroud)
您还可以检索所有文档并使用numpy.diff以返回如下所示的派生:
import numpy as np
for document in collection.find({}, {'time_series': 1}):
result = np.diff(document['time_series'])
Run Code Online (Sandbox Code Playgroud)