fbprophet at scale with applyInPandas resulting unexpected count values [PySpark]

Viewed 87

I am using applyInPandas to implement a forecast function over a sampled data using groupBy on ID. The end goal is to calculate MAPE for each ID.

    
def forecast_balance(history_pd: pd.DataFrame) -> pd.DataFrame:

    anonym_cis = history_pd.at[0,'ID']
    
    # instantiate the model, configure the parameters
    model = Prophet(
        interval_width=0.95,
        growth='linear',
        daily_seasonality=True,
        weekly_seasonality=True,
        yearly_seasonality=False,
        seasonality_mode='multiplicative'
    )

    # fit the model
    model.fit(history_pd)

    # configure predictions
    future_pd = model.make_future_dataframe(
        periods=30,
        freq='d',
        include_history=True
    )

    # make predictions
    results_pd = model.predict(future_pd)
    results_pd.loc[:, 'ID'] = anonym_cis

    # . . .


    # return predictions
    return results_pd[['ds', 'ID', 'yhat', 'yhat_upper', 'yhat_lower']]



results = (
    fr_sample
    .groupBy('ID')
    .applyInPandas(forecast_balance, schema=result_schema)
    )


I am getting am expected predictive results. However, when I count the number of rows for each ID in input data and the output data, it doesn't match. I would like to know from where/how these extra 30 (292-262) rows are getting created in the process for each ID.

+----------+-----+
|        ID|count|
+----------+-----+
|    482726|  262|
|    482769|  262|
|    483946|  262|
|    484124|  262|
|    484364|  262|
|    485103|  262|
+----------+-----+


+----------+-----+
|        ID|count|
+----------+-----+
|    482726|  292|
|    482769|  292|
|    483946|  292|
|    484124|  292|
|    484364|  292|
|    485103|  292|
+----------+-----+

Note: This is how I am calculating MAPE as of now which is not for each ID but a over all data, hence resulting a single value (e.g. 1.4382).

def gr_mape_val(pd_sample_df, result_df):
  result_df = result_df.toPandas()
  actuals_pd = pd_sample_df[pd_sample_df['ds'] < date(2022, 3, 19) ]['y']
  predicted_pd = result_df[ result_df['ds'] < pd.to_datetime('2022-03-19') ]['yhat']
  mape = mean_absolute_percentage_error(actuals_pd, predicted_pd) 
  return mape

To use it in groupBy format for each ID, I need to have both the above mentioned count values matched but I am not able to figure out, how?

1 Answers

I just found what was going on there: Basically with make_future_dataframe, I am creating 30 extra datapoints which was changing the total count of predicted_pd.

This can be simply solved by using df.na.drop()

pd_sample_df.join(result_df, on=['ID', 'ds'], how='outer').na.drop()

Related