2

我正在尝试使用 Dask，特别是 dask 延迟以使用 rpy2 和 R 中的预测包并行生成时间序列预测。我的过程仅在使用 1 个核心时有效，但我得到了

NotImplementedError: Conversion 'py2ri' not defined for objects of type '<class 'pandas.core.series.Series'>'

当使用超过 1 个核心的 dask 延迟时。用于重现此问题的代码如下所示：

from rpy2.robjects.packages import importr
from rpy2.robjects import pandas2ri
import rpy2.robjects as robjects
#get ts object as python object
ts=robjects.r('ts')
pandas2ri.activate()

import pandas as pd
import numpy as np
from dask.distributed import Client, LocalCluster
import dask

#start cluster:

cluster = LocalCluster()
client = Client(cluster)

#define R function to generate time series in R from python series
def r_vecs(time_series):

    rdata=ts(time_series,frequency=12)

    return rdata

#Generate DataFrame of time series
rows = 24
ncolumns = 5
column_names = ['ts1','ts2','ts3','ts4','ts5']
df = pd.DataFrame(np.random.randint(0,10000,size=(rows, ncolumns)), columns=column_names)
df_date_index = pd.date_range(end='2018-04-01', periods=rows, freq='MS')
df.index = df_date_index

使用 dask delay 循环遍历数据帧中的每个时间序列并变成一个时间序列

作品：

output_fc_R = []
for i in df:
    forecasted_series = r_vecs(df[i])
    output_fc_R.append(forecasted_series)

output_fc_R

不起作用：

#Try to forecast in parallel with Dask
output_fc_R = []
for i in df:
    forecasted_series = dask.delayed(r_vecs)(df[i])
    output_fc_R.append(forecasted_series)

total = dask.delayed(output_fc_R).compute()
4

1 回答 1

2

我仍然不确定究竟是什么导致了这个问题，但是当我第一次将时间序列显式转换为 R Intvector 对象时，事情似乎正常工作。

def r_vecs(time_series):

    time_series = robjects.IntVector(time_series)
    rdata=ts(time_series,frequency=12)

    return rdata

在我原来的帖子中，还存在与通过评估 python 字符串在预测包中拟合 R 模型相关的不同问题。如果您想关注完整的线程，请参阅：https ://github.com/dask/distributed/issues/1939

于 2018-04-30T19:57:23.760 回答