Parallel Forecast n_jobs >1

[tuttoaposto]

I was trying out the parallel forecast function and got the following error. Did I miss anything in my code? I ended up modifying core.py to resolve that.

---> 18 forecasts = fcst.forecast(hzn, level=(80, 95))
19 end = time.time()
20 print('Train + Forecast Time:', end-strt)

File ~\python_venv\py395\lib\site-packages\statsforecast\core.py:145, in StatsForecast.forecast(self, h, xreg, level)
140 dates = np.hstack([
141 pd.date_range(last_date + self.freq, periods=h, freq=self.freq)
142 for last_date in self.last_dates
143 ])
144 idx = pd.Index(np.repeat(self.uids, h), name='unique_id')
--> 145 return pd.DataFrame({'ds': dates, **fcsts}, index=idx)

File ~\python_venv\py395\lib\site-packages\pandas\core\frame.py:636, in DataFrame.init(self, data, index, columns, dtype, copy)
630 mgr = self._init_mgr(
631 data, axes={"index": index, "columns": columns}, dtype=dtype, copy=copy
632 )
634 elif isinstance(data, dict):
635 # GH#38939 de facto copy defaults to False only in non-dict cases
--> 636 mgr = dict_to_mgr(data, index, columns, dtype=dtype, copy=copy, typ=manager)
637 elif isinstance(data, ma.MaskedArray):
638 import numpy.ma.mrecords as mrecords

File ~\python_venv\py395\lib\site-packages\pandas\core\internals\construction.py:502, in dict_to_mgr(data, index, columns, dtype, typ, copy)
494 arrays = [
495 x
496 if not hasattr(x, "dtype") or not isinstance(x.dtype, ExtensionDtype)
497 else x.copy()
498 for x in arrays
499 ]
500 # TODO: can we get rid of the dt64tz special case above?
--> 502 return arrays_to_mgr(arrays, columns, index, dtype=dtype, typ=typ, consolidate=copy)

File ~\python_venv\py395\lib\site-packages\pandas\core\internals\construction.py:125, in arrays_to_mgr(arrays, columns, index, dtype, verify_integrity, typ, consolidate)
122 index = ensure_index(index)
124 # don't force copy because getting jammed in an ndarray anyway
--> 125 arrays = _homogenize(arrays, index, dtype)
126 # _homogenize ensures
127 # - all(len(x) == len(index) for x in arrays)
128 # - all(x.ndim == 1 for x in arrays)
(...)
131
132 else:
133 index = ensure_index(index)

File ~\python_venv\py395\lib\site-packages\pandas\core\internals\construction.py:628, in _homogenize(data, index, dtype)
623 val = lib.fast_multiget(val, oindex._values, default=np.nan)
625 val = sanitize_array(
626 val, index, dtype=dtype, copy=False, raise_cast_failure=False
627 )
--> 628 com.require_length_match(val, index)
630 homogenized.append(val)
632 return homogenized

File ~\python_venv\py395\lib\site-packages\pandas\core\common.py:557, in require_length_match(data, index)
553 """
554 Check the length of data matches the length of the index.
555 """
556 if len(data) != len(index):
--> 557 raise ValueError(
558 "Length of values "
559 f"({len(data)}) "
560 "does not match length of index "
561 f"({len(index)})"
562 )

ValueError: Length of values (4) does not match length of index (96)

This is my code. Training set is assembled from two time series, each has its own unique_id. I want to forecast out for 48 steps:

df1.index = pd.Index(['df1']*len(df1), name = 'unique_id')
df2.index = pd.Index(['df2']*len(df2), name = 'unique_id')
train = pd.concat([df1, df2])
train
	unique_id	ds			y
	df1		2022-02-16 13:00:00	141
	df1		2022-02-16 14:00:00	142
	df1		2022-02-16 15:00:00	144
	df1		2022-02-16 16:00:00	143
	df1		2022-02-16 17:00:00	141
	...	...	...
	df2		2022-03-02 09:00:00	202
	df2		2022-03-02 10:00:00	202
	df2		2022-03-02 11:00:00	203
	df2		2022-03-02 12:00:00	201
	df2		2022-03-02 13:00:00	198

fcst = StatsForecast(
    train, 
    models=[(auto_arima, 24)], 
    freq='H', 
    n_jobs=2
)
forecasts = fcst.forecast(h = 48, level=(80, 95))

The error happened when the intermediate step didn't output a dict. Instead it output an array of a list of forecast arrays alternating with column headers like this:

array([forecast-array-for df1, list(['mean', 'lo-95', 'hi-95']),
       forecast-array-for df2, list(['mean', 'lo-95', 'hi-95'])]).

And pd.DataFrame() couldn't resolve it.

So I made the following changes in core.py the line for `model_forecasts' as a work-around:

def _data_parallel_forecast(self, h, xreg, level):
        fcsts = {}
        logger.info('Computing forecasts')
        gas = self.ga.split(self.n_jobs)
        with ProcessPoolExecutor(self.n_jobs) as executor:
            for model_args in self.models:
                model, *args = _as_tuple(model_args)
                model_name = _build_forecast_name(model, *args)
                futures = []
                for ga in gas:
                    future = executor.submit(ga.compute_forecasts, h, model, xreg, level, *args)
                    futures.append(future)
                

>            #--- Replace this line with the followings:  -----------------
>                 #model_forecasts = np.hstack([f.result() for f in futures])
>                 values = np.vstack([f.result()[0] for f in futures])
>                 values = np.transpose(values)
>                 keys = futures[0].result()[1]
>                 model_forecasts = dict([(k, v) for k, v in zip(keys, values)])
>           #-------------------------------------------------------------

                if isinstance(model_forecasts, dict):
                    for k, v in model_forecasts.items():
                        fcsts[f'{model_name}_{k}'] = v
                else:
                    fcsts[model_name] = model_forecasts

Finally I was able to get a dataframe with 80% & 95% prediction intervals or just the 95% with
forecasts = fcst.forecast(hzn, level=(95,)) or
forecasts = fcst.forecast(hzn, level=[95]).

unique_id	ds			auto_arima_season_length-24_mean	auto_arima_season_length-24_lo-95	auto_arima_season_length-24_hi-95
			
df1		2022-03-02 14:00:00	140.808182				136.821686				144.794678
df1		2022-03-02 15:00:00	139.969681				134.331924				145.607437
df1		2022-03-02 16:00:00	137.388947				130.484131				144.293762
df1		2022-03-02 17:00:00	135.294113				127.321121				143.267105

Let me know if there is a better way of handling this. Thank you!

[AzulGarza]

Hi @tuttoaposto! It seems that the problem is related to #63. Yesterday we released a version that fixes the issue. Could you please try with the new version using pip install --upgrade statsforecast or pip install statsforecast>=0.5.1? Let us know if the problem is solved or if you have more problems. :)

[tuttoaposto]

Thank you for your prompt response!

I only skimmed through the open issues. I will check the closed tab next
time. Thank you for the reminder!

I will try your updated version and test parallel run again when I return
to work.

[mergenthaler]

@tuttoaposto, if you are open, we would like to connect with you. You can send me an email or we can talk via our Slack.

[tuttoaposto]

Hi @mergenthaler,

Happy to connect! Thank you for the invite.