我正在尝试使用 dask 在 python 中并行化时间序列预测。数据的格式是每个时间序列都是一列,并且它们具有共同的每月日期索引。我有一个自定义预测函数,它返回一个带有拟合值和预测值的时间序列对象。我想将此函数应用于数据帧(所有时间序列)的所有列,并返回一个新的数据帧,其中所有这些序列都要上传到数据库。 我已经通过运行使代码可以工作:
data = pandas_df.copy()
ddata = dd.from_pandas(data, npartitions=1)
res = ddata.map_partitions(lambda df: df.apply(forecast_func,
axis=0)).compute(get=dask.multiprocessing.get)
我的问题是,Dask 中是否有一种方法可以按列而不是行进行分区,因为在这个用例中,我需要保持有序时间索引,以便预测功能正常工作。
如果没有,我将如何重新格式化数据以允许高效的大规模预测,并且仍然以我需要的格式返回数据,然后推送到数据库?
我使用了 dask.delayed 解决方案,它运行得非常好,大约只需要使用本地集群 1/3 的时间。
对于任何对我实施的解决方案感兴趣的人:
from dask.distributed import Client, LocalCluster
import pandas as pd
import dask
cluster = LocalCluster(n_workers=3,ncores=3)
client = Client(cluster)
#get list of time series back
output = []
for i in small_df:
forecasted_series = dask.delayed(custom_forecast_func)(small_df[i])
output.append(forecasted_series)
total = dask.delayed(output).compute()
#combine list of series into 1 dataframe
full_df = pd.concat(total,ignore_index=False,keys=small_df.columns,names=['time_series_names','Date'])
final_df = full_df.to_frame().reset_index()
final_df.columns = ['time_series_names','Date','value_variable']
final_df.head()
这为您提供了融化的数据框结构,因此如果您希望该系列成为列,您可以使用它进行转换
pivoted_df = final_df.pivot(index='Date', columns='time_series_names', values='value_variable')
Dask dataframe 仅按行对数据进行分区。 请参阅 Dask 数据框文档
然而,Dask 数组可以沿任何维度进行分区。 你可以使用 Numpy 语义而不是 Pandas 语义。