将季节性分解的趋势应用到dask DataFrame的每一列，Python

Question

正如标题所说，我无法运行这段代码。

def simple_map(x):
    y = seasonal_decompose(x,model='additive',extrapolate_trend='freq',period=7,two_sided=False)
    return y.trend

b.map_partitions(simple_map,meta=b).compute()

其中b是一个dask DataFrame，以datetime为索引，以float为列， seasonal_decompose是statsmodel的。

这是我得到的结果。

Index(...) must be called with a collection of some kind, 'seasonal' was passed

如果我这样做：

b.apply(simple_map,axis=0)

其中b是一个pandas DataFrame，我得到我想要的东西。

我错在哪里？

#

可复制的例子。

import pandas as pd
from statsmodels.tsa.seasonal import seasonal_decompose

d = {'Val1': [3, 2,7,5], 'Val2': [2, 4,8,6]}
b=pd.DataFrame(data=d)
b=b.set_index(pd.to_datetime(['25/12/1991','26/12/1991','27/12/1991','28/12/1991']))

def simple_map(x):
    y =seasonal_decompose(x,model='additive',extrapolate_trend='freq',period=2,two_sided=False)
    return y.trend

b.apply(simple_map,axis=0)

            Val1    Val2
1991-12-25  0.70    0.9
1991-12-26  2.10    2.7
1991-12-27  3.50    4.5
1991-12-28  5.25    6.5

这是我想用dask做的事，但我不能。

事实上。

import dask.dataframe as dd

c=dd.from_pandas(b, npartitions=1)
c.map_partitions(simple_map,meta=c).compute()

产生上述指定的错误。

Answer 1

谢谢你的例子

从apply的docstring中可以看到

传递给函数的对象是Series对象，其索引是DataFrame的索引(axis=0)

然而， map_partitions 是要在整个Dataframe上工作。 我建议稍微重写一下这个函数。

 def simple_map_2(x):
     xVal1 = seasonal_decompose(x.Val1,model='additive',extrapolate_trend='freq',period=2,two_sided=False)
     xVal2 = seasonal_decompose(x.Val2,model='additive',extrapolate_trend='freq',period=2,two_sided=False)
     return pd.DataFrame({'Val1': xVal1.trend, 'Val2': xVal2.trend})

c.map_partitions(simple_map_2,meta=make_meta(c)).compute()

            Val1  Val2
1991-12-25  0.70   0.9
1991-12-26  2.10   2.7
1991-12-27  3.50   4.5
1991-12-28  5.25   6.5