python-如何使用dask有效地并行化时间序列预测？

我正在尝试使用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中是否有一种方法可以按列而不是按行进行分区,因为在此用例中,我需要保持有序时间索引不变,以使预测功能正常工作.

如果没有,我将如何重新格式化数据以实现高效的大规模预测,并仍然以我需要的格式返回数据,然后将其推送到数据库？

example of data format

解决方法:

感谢您的帮助,我真的很感激.我使用了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')

small_df is in this format in pandas dataframe with Date being the index

标签：dask,parallel-processing,time-series,forecasting,python
来源： https://codeday.me/bug/20191025/1927437.html