[FugueBackend: Dask Distributed] Result is empty when using remote dask cluster

[ibyter]

What happened + What you expected to happen

I wanted to try distributed computation using dask. I followed the docs https://nixtla.github.io/statsforecast/distributed.fugue.html and it's working fine using dask local client.

However, when I attempted to use a remote dask cluster, I got empty result:

from dask.distributed import Client
from fugue_dask import DaskExecutionEngine
from statsforecast import StatsForecast
from statsforecast.models import Naive
from statsforecast.utils import generate_series
from statsforecast.distributed.fugue import FugueBackend
import pandas as pd

# Instantiate FugueBackend with DaskExecutionEngine
dask_client = Client('tcp://***:***')
engine = DaskExecutionEngine(dask_client=dask_client)
remote_backend = FugueBackend(engine=engine, as_local=True)
Y_df = pd.read_parquet('https://datasets-nixtla.s3.amazonaws.com/m4-hourly.parquet')

from statsforecast.models import (
    AutoARIMA,
    HoltWinters,
    CrostonClassic as Croston, 
    HistoricAverage,
    DynamicOptimizedTheta as DOT,
    SeasonalNaive
)

# Create a list of models and instantiation parameters
models = [
    AutoARIMA(season_length=24),
    HoltWinters(),
    Croston(),
    SeasonalNaive(season_length=24),
    HistoricAverage(),
    DOT(season_length=24)
]

# Instantiate StatsForecast class as sf
sf = StatsForecast(
    df=Y_df, 
    models=models,
    freq='H', 
    n_jobs=-1,
    fallback_model = SeasonalNaive(season_length=7),
    backend=remote_backend
)

forecasts_df = sf.forecast(h=48, level=[90])
print(forecasts_df.size) # returns 0

I didn't get any errors, but the result is empty

Versions / Dependencies

Local setup:

OS: mac Python: 3.10 py packages:

• dask==2022.12.1
• dask-cloudprovider==2022.10.0
• datasetsforecast==0.0.7
• distributed==2022.12.1
• fugue==0.7.3
• fugue-sql-antlr==0.1.1
• s3fs==2022.11.0
• statsforecast==1.4.0
• pandas==1.5.2

Dask cluster:

• scheduer & worker: dask, version 2022.12.1
• Python: 3.8

Reproduction script

• Create a remote dask cluster
• Use the code above to connect to dask scheduler, and run the sample

Issue Severity

High: It blocks me from completing my task.

[ibyter]

@goodwanghan @FedericoGarza

[goodwanghan]

@ibyter sorry I just saw this, will investigate

[kvnkho]

I'll take this issue