Невозможно разместить фрейм данных с помощью fbprophet, используя dask для чтения csv в фрейм данных

К сожалению, сегодня Prophet не поддерживает фреймы данных Dask.

Пример, на который вы ссылаетесь, показывает использование Dask для ускорения установки Prophet на фреймы данных Pandas. Dask Dataframe - это только один из способов использования Dask.

Как уже было предложено , можно использовать dask.delayed с пандами DataFrame и пропустить dask.dataframe.

Вы можете использовать упрощенную версию конвейера _4 _-_ 5 _-_ 6_ показан для пользовательских вычислений с использованием Dask.

Вот один из возможных подходов, основанный на этом типе настраиваемого конвейера с использованием небольшого набора данных (для создания MCVE) - каждый шаг в конвейере будет задерживаться.

Импорт

import numpy as np
import pandas as pd
from dask import delayed
from dask.distributed import Client
from fbprophet import Prophet

Создайте некоторые данные в .csv с именами столбцов Time (UTC), a и b

def generate_csv(nrows, fname):
    df = pd.DataFrame(np.random.rand(nrows, 2), columns=["a", "b"])
    df["Time (UTC)"] = pd.date_range(start="1850-01-01", periods=nrows)
    df.to_csv(fname, index=False)

Сначала напишите функцию load из конвейера, чтобы загрузить .csv с помощью Pandas и отложить ее выполнение, используя _ 15_ декоратор

• было бы хорошо использовать read_csv с nrows , чтобы увидеть, как конвейер работает с подмножеством данных, а не загружает их все.
• это вернет объект dask.delayed, а не pandas.DataFrame

@delayed
def load_data(fname, nrows=None):
    return pd.read_csv(fname, nrows=nrows)

Теперь создайте функцию process, чтобы обрабатывать данные с помощью pandas, снова с задержкой, поскольку ее ввод - это объект dask.delayed, а не pandas.DataFrame

@delayed
def process_data(df):
    df = df.rename(columns={"Time (UTC)": "ds"})
    df["y"] = df[["a", "b"]].mean(axis=1)
    return df

Последняя функция - она ​​будет тренировать fbprophet на данных (загруженных из .csv и обработанных, но с задержкой) для составления прогноза. Эта analyze функция также задерживается, так как один из ее входов является dask.delayed объектом.

@delayed
def analyze(df, horizon):
    m = Prophet(daily_seasonality=True)
    m.fit(df)
    future = m.make_future_dataframe(periods=horizon)
    forecast = m.predict(future)
    return forecast

Запустите конвейер (при запуске из сценария Python требуется __name__ == "__main__")

• the output of the pipeline (a forecast by fbprophet) is stored in a variable result, which is delayed
  • when this output is computed, this will generate a pandas.DataFrame (corresponding to the output of a forecast by fbprophet), so it can be evaluated using result.compute()

if __name__ == "__main__":
    horizon = 8
    num_rows_data = 40
    num_rows_to_load = 35
    csv_fname = "my_file.csv"

    generate_csv(num_rows_data, csv_fname)

    client = Client()  # modify this as required

    df = load_data(csv_fname, nrows=num_rows_to_load)
    df = process_data(df)
    result = analyze(df, horizon)
    forecast = result.compute()

    client.close()

    assert len(forecast) == num_rows_to_load + horizon
    print(forecast[["ds", "yhat", "yhat_lower", "yhat_upper"]].head())

Выход

          ds      yhat  yhat_lower  yhat_upper
0 1850-01-01  0.330649    0.095788    0.573378
1 1850-01-02  0.493025    0.266692    0.724632
2 1850-01-03  0.573344    0.348953    0.822692
3 1850-01-04  0.491388    0.246458    0.712400
4 1850-01-05  0.307939    0.066030    0.548981