python - 如何将 Dask Delayed 与 rpy2 一起使用？

Question

我正在尝试使用 Dask，特别是 dask delayed 以使用 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 delayed循环遍历dataframe中的每一个时间序列，变成一个时间序列

作品:

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()

Answer 1

我仍然不确定到底是什么导致了这个问题，但是当我第一次明确地将时间序列转换为 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