{ "cells": [ { "attachments": {}, "cell_type": "markdown", "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": {}, "inputWidgets": {}, "nuid": "a02bb3a2-e5d7-4d14-b966-827457675b75", "showTitle": false, "title": "" } }, "source": [ "# Retail Demand Forecasting\n", "\n", "Retailers often deal with large datasets that exceed the memory capacity of a single machine. \n", "These datasets also exhibit diverse characteristics, particularly when considering the varying behaviors of different stores. \n", "Consequently, data scientists face several challenges, including:\n", "\n", "- Limitations of Pandas/Python in handling big datasets, necessitating the use of Pyspark.\n", "- The need to create separate models for different groups in the dataset (e.g., stores or items) to optimize performance.\n", "- Despite the dataset's overall size, each group (items/stores) is small enough to leverage models similar to those provided by ``scikit-learn``.\n", "\n", "This tutorial demonstrates how data scientists can overcome these challenges using ``ForecastFlowML``.\n", "\n", "## Situation\n", "\n", "Given the considerable size of our sample dataset, it exceeds the memory capacity for direct loading.\n", " Consequently, we must leverage PySpark, a distributed processing framework, to effectively manage the data processing operations.\n", "\n", "Fortunately, within this extensive dataset, we possess store-specific information.\n", "The individual store data, being relatively smaller in size, can comfortably fit within the memory of a single machine. \n", "Consequently, we can leverage this advantage to employ ``scikit-learn`` for model creation, without encountering any memory limitations.\n", "\n", "## Goal\n", "\n", "- Develop independent models for each store in the dataset.\n", "- Parallelize the feature engineering, training, and inference steps.\n", "- Employ ``LightGBM`` as the machine learning algorithm.\n", "- Utilize a direct multi-step forecasting approach.\n", "- Conduct backtesting to assess the model's performance." ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "## Import Packages" ] }, { "cell_type": "code", "execution_count": 38, "metadata": {}, "outputs": [], "source": [ "from forecastflowml import ForecastFlowML\n", "from forecastflowml import FeatureExtractor\n", "from forecastflowml.data.loader import load_walmart_m5\n", "from lightgbm import LGBMRegressor\n", "from pyspark.sql import SparkSession\n", "import pyspark.sql.functions as F\n", "import plotly.express as px\n", "import plotly.io as pio\n", "import pandas as pd\n", "import sys\n", "import os\n", "\n", "os.environ[\"PYSPARK_PYTHON\"] = sys.executable\n", "pd.set_option(\"display.max_columns\", 40)" ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "## Initialize Spark" ] }, { "cell_type": "code", "execution_count": 39, "metadata": {}, "outputs": [], "source": [ "spark = (\n", " SparkSession.builder.master(\"local[4]\")\n", " .config(\"spark.driver.memory\", \"4g\")\n", " .config(\"spark.sql.shuffle.partitions\", \"4\")\n", " .config(\"spark.sql.execution.pyarrow.enabled\", \"true\")\n", " .getOrCreate()\n", ")" ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "## Sample Dataset" ] }, { "cell_type": "code", "execution_count": 40, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+--------------------+-----------+-------+------+--------+--------+----------+-----+\n", "| id| item_id|dept_id|cat_id|store_id|state_id| date|sales|\n", "+--------------------+-----------+-------+------+--------+--------+----------+-----+\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-15| 3.0|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-16| 0.0|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-17| 1.0|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-18| 0.0|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-19| 0.0|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-20| 0.0|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-21| 0.0|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-22| 0.0|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-23| 0.0|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-24| 0.0|\n", "+--------------------+-----------+-------+------+--------+--------+----------+-----+\n", "only showing top 10 rows\n", "\n" ] } ], "source": [ "df = load_walmart_m5(spark).localCheckpoint()\n", "df.show(10)" ] }, { "cell_type": "code", "execution_count": 41, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+--------+\n", "|store_id|\n", "+--------+\n", "| TX_1|\n", "| WI_1|\n", "| CA_1|\n", "| TX_2|\n", "+--------+\n", "\n" ] } ], "source": [ "df.select(\"store_id\").dropDuplicates().show()" ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "## Approach" ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "There are 4 different stores in our dataset. Since each store might have different behaviours, \n", "it would be better to build an independent model for each of them. Moreover,\n", "there will be 4 models per store that is going to predict 7 days. \n", "In total, we are going to have 4 x 4 = 16 models.\n", "\n", "\n", "![image info](../_static/store_based_models.svg)" ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "As we are going to build 4 weekly independent direct forecasting models, each of them are going to use different lag features \n", "based upon their forecasting horizon.\n", "\n", "![image info](../_static/lag.svg)" ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "## Feature Engineering\n", "\n", "Let's create the features that will be used for forecasting. \n", "Our package has a seperate module ``FeatureExtractor`` to extract features.\n", "Spesifically, we are going to extract:\n", "\n", "- Lags\n", "- Roling means\n", "- Date features\n", "- Stockout features (counting consecutive no sales periods)\n", "- History length (number of periods passed after the start of the time series)" ] }, { "cell_type": "code", "execution_count": 42, "metadata": {}, "outputs": [], "source": [ "feature_extractor = FeatureExtractor(\n", " id_col=\"id\",\n", " date_col=\"date\",\n", " target_col=\"sales\",\n", " lag_window_features={\n", " \"lag\": [7 * (i + 1) for i in range(7)],\n", " \"mean\": [[window, lag] for lag in [7, 14, 21, 28] for window in [7, 14, 30]],\n", " },\n", " date_features=[\n", " \"day_of_month\",\n", " \"day_of_week\",\n", " \"week_of_year\",\n", " \"quarter\",\n", " \"month\",\n", " \"year\",\n", " ],\n", " count_consecutive_values={\n", " \"value\": 0,\n", " \"lags\": [7, 14, 21, 28],\n", " },\n", " history_length=True,\n", ")" ] }, { "cell_type": "code", "execution_count": 43, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+--------------------+-----------+-------+------+--------+--------+----------+-----+-----+------+------+------+------+------+------+-------------------+--------------------+--------------------+--------------------+---------------------+---------------------+--------------------+---------------------+---------------------+--------------------+---------------------+---------------------+-----------------------------+------------------------------+------------------------------+------------------------------+--------------+------------+-----------+------------+-------+-----+----+\n", "| id| item_id|dept_id|cat_id|store_id|state_id| date|sales|lag_7|lag_14|lag_21|lag_28|lag_35|lag_42|lag_49|window_7_lag_7_mean|window_14_lag_7_mean|window_30_lag_7_mean|window_7_lag_14_mean|window_14_lag_14_mean|window_30_lag_14_mean|window_7_lag_21_mean|window_14_lag_21_mean|window_30_lag_21_mean|window_7_lag_28_mean|window_14_lag_28_mean|window_30_lag_28_mean|count_consecutive_value_lag_7|count_consecutive_value_lag_14|count_consecutive_value_lag_21|count_consecutive_value_lag_28|history_length|day_of_month|day_of_week|week_of_year|quarter|month|year|\n", "+--------------------+-----------+-------+------+--------+--------+----------+-----+-----+------+------+------+------+------+------+-------------------+--------------------+--------------------+--------------------+---------------------+---------------------+--------------------+---------------------+---------------------+--------------------+---------------------+---------------------+-----------------------------+------------------------------+------------------------------+------------------------------+--------------+------------+-----------+------------+-------+-----+----+\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-15| 3.0| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| 1| 15| 4| 3| 1| 1|2015|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-16| 0.0| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| 2| 16| 5| 3| 1| 1|2015|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-17| 1.0| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| 3| 17| 6| 3| 1| 1|2015|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-18| 0.0| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| 4| 18| 7| 3| 1| 1|2015|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-19| 0.0| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| 5| 19| 1| 4| 1| 1|2015|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-20| 0.0| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| 6| 20| 2| 4| 1| 1|2015|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-21| 0.0| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| 7| 21| 3| 4| 1| 1|2015|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-22| 0.0| 3.0| null| null| null| null| null| null| 3.0| 3.0| 3.0| null| null| null| null| null| null| null| null| null| 0| null| null| null| 8| 22| 4| 4| 1| 1|2015|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-23| 0.0| 0.0| null| null| null| null| null| null| 1.5| 1.5| 1.5| null| null| null| null| null| null| null| null| null| 1| null| null| null| 9| 23| 5| 4| 1| 1|2015|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-24| 0.0| 1.0| null| null| null| null| null| null| 1.3333333333333333| 1.3333333333333333| 1.3333333333333333| null| null| null| null| null| null| null| null| null| 0| null| null| null| 10| 24| 6| 4| 1| 1|2015|\n", "+--------------------+-----------+-------+------+--------+--------+----------+-----+-----+------+------+------+------+------+------+-------------------+--------------------+--------------------+--------------------+---------------------+---------------------+--------------------+---------------------+---------------------+--------------------+---------------------+---------------------+-----------------------------+------------------------------+------------------------------+------------------------------+--------------+------------+-----------+------------+-------+-----+----+\n", "only showing top 10 rows\n", "\n" ] } ], "source": [ "df_features = feature_extractor.transform(df).localCheckpoint()\n", "df_features.show(10)" ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "## Training/Test Dataset" ] }, { "cell_type": "code", "execution_count": 44, "metadata": {}, "outputs": [], "source": [ "df_train = df_features.filter(F.col(\"date\") < \"2016-04-25\")\n", "df_test = df_features.filter(F.col(\"date\") >= \"2016-04-25\")" ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "## Initialize ForecastFlowML" ] }, { "cell_type": "code", "execution_count": 45, "metadata": {}, "outputs": [], "source": [ "forecast_flow = ForecastFlowML(\n", " group_col=\"store_id\", # column to slice the dataframe\n", " id_col=\"id\", # time series identifier column\n", " date_col=\"date\", # date column\n", " target_col=\"sales\", # target column\n", " categorical_cols=[\n", " \"item_id\",\n", " \"dept_id\",\n", " \"cat_id\",\n", " ], # columns to be treated as categorical\n", " date_frequency=\"days\", # dataset date frequency\n", " model_horizon=7, # single model predicts 7 days\n", " max_forecast_horizon=28, # total forecast horizon will be 28 days\n", " model=LGBMRegressor(), # algorithm\n", " use_lag_range=21, # allow to use 21 days of extra lag features to use based upon the forecast horizon\n", ")" ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "## Cross Validation\n", "\n", "First, let's perform some backtesting to understand the model performance. We are going to \n", "predict the last 3x28 days." ] }, { "cell_type": "code", "execution_count": 46, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+--------+--------------------+-------------------+---+-----+----------+\n", "|store_id| id| date| cv|sales|prediction|\n", "+--------+--------------------+-------------------+---+-----+----------+\n", "| CA_1|FOODS_1_064_CA_1_...|2016-03-28 00:00:00| 0| 2.0| 0.979212|\n", "| CA_1|FOODS_1_064_CA_1_...|2016-03-29 00:00:00| 0| 0.0| 1.047827|\n", "| CA_1|FOODS_1_064_CA_1_...|2016-03-30 00:00:00| 0| 0.0|0.92554504|\n", "| CA_1|FOODS_1_064_CA_1_...|2016-03-31 00:00:00| 0| 0.0| 0.9589823|\n", "| CA_1|FOODS_1_064_CA_1_...|2016-04-01 00:00:00| 0| 3.0| 1.305723|\n", "| CA_1|FOODS_1_064_CA_1_...|2016-04-02 00:00:00| 0| 0.0| 1.5510178|\n", "| CA_1|FOODS_1_064_CA_1_...|2016-04-03 00:00:00| 0| 3.0| 1.4713657|\n", "| CA_1|FOODS_1_121_CA_1_...|2016-03-28 00:00:00| 0| 0.0| 0.560704|\n", "| CA_1|FOODS_1_121_CA_1_...|2016-03-29 00:00:00| 0| 0.0| 0.5866643|\n", "| CA_1|FOODS_1_121_CA_1_...|2016-03-30 00:00:00| 0| 0.0| 0.5995418|\n", "+--------+--------------------+-------------------+---+-----+----------+\n", "only showing top 10 rows\n", "\n" ] } ], "source": [ "cv_forecast = forecast_flow.cross_validate(df_train, n_cv_splits=3).localCheckpoint()\n", "cv_forecast.show(10)" ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "### Calculate RMSE" ] }, { "cell_type": "code", "execution_count": 47, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+---+------------------+\n", "| cv| rmse|\n", "+---+------------------+\n", "| 0|1.9499041165007749|\n", "| 1|2.0947956593876733|\n", "| 2| 1.925051121492435|\n", "+---+------------------+\n", "\n" ] } ], "source": [ "(\n", " cv_forecast.withColumn(\n", " \"error_squared\", F.pow(F.abs(F.col(\"sales\") - F.col(\"prediction\")), 2)\n", " )\n", " .groupBy(\"cv\")\n", " .agg(F.pow(F.avg(\"error_squared\"), 0.5).alias(\"rmse\"))\n", " .orderBy(\"cv\")\n", " .show()\n", ")" ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "### Visualize Cross Validation Predictions" ] }, { "cell_type": "code", "execution_count": 48, "metadata": {}, "outputs": [], "source": [ "cv_past_future = (\n", " df_train.select(\"id\", \"store_id\", \"date\", \"sales\")\n", " .join(\n", " cv_forecast.select(\"id\", \"date\", \"cv\", \"prediction\"),\n", " on=[\"id\", \"date\"],\n", " how=\"left\",\n", " )\n", " .groupBy(\"id\", \"store_id\", \"date\", \"sales\")\n", " .pivot(\"cv\")\n", " .sum(\"prediction\")\n", " .groupBy(\"store_id\", \"date\")\n", " .agg(\n", " F.sum(\"sales\").alias(\"sales\"),\n", " *[F.sum(f\"{i}\").alias(f\"cv_{i}\") for i in range(3)],\n", " )\n", " .orderBy(\"store_id\", \"date\")\n", ").toPandas()" ] }, { "cell_type": "code", "execution_count": 49, "metadata": {}, "outputs": [ { "data": { "text/html": [ " \n", " " ] }, "metadata": {}, "output_type": "display_data" }, { "data": { "text/html": [ "
" ] }, "metadata": {}, "output_type": "display_data" } ], "source": [ "pio.renderers.default = \"notebook\"\n", "fig = px.line(\n", " cv_past_future,\n", " x=\"date\",\n", " y=[\"sales\", *[f\"cv_{i}\" for i in range(3)]],\n", " facet_row_spacing=0.04,\n", " facet_col=\"store_id\",\n", " facet_col_wrap=2,\n", " height=700,\n", " width=720,\n", ")\n", "fig.update_layout(\n", " legend=dict(orientation=\"h\", yanchor=\"top\", y=1.09, xanchor=\"center\", x=0.5),\n", " margin=dict(l=0, r=10, t=5, b=5),\n", " legend_title=\"\",\n", ")\n", "fig.update_traces(line=dict(width=1.7))\n", "fig.update_yaxes(matches=None, title=\"\")\n", "fig.update_xaxes(type=\"date\", range=[\"2015-11-01\", \"2016-04-24\"])" ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "## Training\n", "\n", "Here we know that the trained models will be small enough to fit into the memory, therefore; we use the ``local_result=True`` which will keep the trained models as an attribute. However, if the trained models are expected to not fit into the memory, then ``local_result=False`` (default parameter) will return models in PySpark DataFrame. " ] }, { "cell_type": "code", "execution_count": 50, "metadata": {}, "outputs": [ { "data": { "text/html": [ "
\n", "\n", "\n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", "
store_idforecast_horizonmodelstart_timeend_timeelapsed_seconds
0CA_1[[1, 2, 3, 4, 5, 6, 7], [8, 9, 10, 11, 12, 13,...[[128, 4, 149, 236, 1, 0, 0, 0, 0, 0, 0, 140, ...19-May-2023 (23:41:17)19-May-2023 (23:41:18)0.6
1TX_1[[1, 2, 3, 4, 5, 6, 7], [8, 9, 10, 11, 12, 13,...[[128, 4, 149, 236, 1, 0, 0, 0, 0, 0, 0, 140, ...19-May-2023 (23:41:18)19-May-2023 (23:41:19)0.6
2WI_1[[1, 2, 3, 4, 5, 6, 7], [8, 9, 10, 11, 12, 13,...[[128, 4, 149, 236, 1, 0, 0, 0, 0, 0, 0, 140, ...19-May-2023 (23:41:19)19-May-2023 (23:41:19)0.5
3TX_2[[1, 2, 3, 4, 5, 6, 7], [8, 9, 10, 11, 12, 13,...[[128, 4, 149, 236, 1, 0, 0, 0, 0, 0, 0, 140, ...19-May-2023 (23:41:16)19-May-2023 (23:41:17)0.7
\n", "
" ], "text/plain": [ " store_id forecast_horizon \\\n", "0 CA_1 [[1, 2, 3, 4, 5, 6, 7], [8, 9, 10, 11, 12, 13,... \n", "1 TX_1 [[1, 2, 3, 4, 5, 6, 7], [8, 9, 10, 11, 12, 13,... \n", "2 WI_1 [[1, 2, 3, 4, 5, 6, 7], [8, 9, 10, 11, 12, 13,... \n", "3 TX_2 [[1, 2, 3, 4, 5, 6, 7], [8, 9, 10, 11, 12, 13,... \n", "\n", " model start_time \\\n", "0 [[128, 4, 149, 236, 1, 0, 0, 0, 0, 0, 0, 140, ... 19-May-2023 (23:41:17) \n", "1 [[128, 4, 149, 236, 1, 0, 0, 0, 0, 0, 0, 140, ... 19-May-2023 (23:41:18) \n", "2 [[128, 4, 149, 236, 1, 0, 0, 0, 0, 0, 0, 140, ... 19-May-2023 (23:41:19) \n", "3 [[128, 4, 149, 236, 1, 0, 0, 0, 0, 0, 0, 140, ... 19-May-2023 (23:41:16) \n", "\n", " end_time elapsed_seconds \n", "0 19-May-2023 (23:41:18) 0.6 \n", "1 19-May-2023 (23:41:19) 0.6 \n", "2 19-May-2023 (23:41:19) 0.5 \n", "3 19-May-2023 (23:41:17) 0.7 " ] }, "execution_count": 50, "metadata": {}, "output_type": "execute_result" } ], "source": [ "forecast_flow.train(df_train, local_result=True)\n", "forecast_flow.model_" ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "## Prediction" ] }, { "cell_type": "code", "execution_count": 51, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+--------+--------------------+-------------------+----------+\n", "|store_id| id| date|prediction|\n", "+--------+--------------------+-------------------+----------+\n", "| CA_1|FOODS_1_064_CA_1_...|2016-04-25 00:00:00| 1.1891526|\n", "| CA_1|FOODS_1_064_CA_1_...|2016-04-26 00:00:00| 1.1587629|\n", "| CA_1|FOODS_1_064_CA_1_...|2016-04-27 00:00:00| 1.1314319|\n", "| CA_1|FOODS_1_064_CA_1_...|2016-04-28 00:00:00| 1.1495334|\n", "| CA_1|FOODS_1_064_CA_1_...|2016-04-29 00:00:00| 1.311355|\n", "| CA_1|FOODS_1_064_CA_1_...|2016-04-30 00:00:00| 1.2894475|\n", "| CA_1|FOODS_1_064_CA_1_...|2016-05-01 00:00:00| 1.375367|\n", "| CA_1|FOODS_1_121_CA_1_...|2016-04-25 00:00:00|0.63143337|\n", "| CA_1|FOODS_1_121_CA_1_...|2016-04-26 00:00:00| 0.5990241|\n", "| CA_1|FOODS_1_121_CA_1_...|2016-04-27 00:00:00| 0.6636088|\n", "+--------+--------------------+-------------------+----------+\n", "only showing top 10 rows\n", "\n" ] } ], "source": [ "forecast = forecast_flow.predict(df_test, spark=spark).localCheckpoint()\n", "forecast.show(10)" ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "### Visualize Predictions" ] }, { "cell_type": "code", "execution_count": 52, "metadata": {}, "outputs": [], "source": [ "past_future = (\n", " df.select(\"id\", \"store_id\", \"date\", \"sales\")\n", " .join(forecast.select(\"id\", \"date\", \"prediction\"), on=[\"id\", \"date\"], how=\"left\")\n", " .groupBy(\"store_id\", \"date\")\n", " .agg(\n", " F.sum(\"sales\").alias(\"sales\"),\n", " F.sum(\"prediction\").alias(\"prediction\"),\n", " )\n", " .orderBy(\"store_id\", \"date\")\n", " .toPandas()\n", ")" ] }, { "cell_type": "code", "execution_count": 53, "metadata": {}, "outputs": [ { "data": { "text/html": [ " \n", " " ] }, "metadata": {}, "output_type": "display_data" }, { "data": { "text/html": [ "
" ] }, "metadata": {}, "output_type": "display_data" } ], "source": [ "pio.renderers.default = \"notebook\"\n", "fig = px.line(\n", " past_future,\n", " x=\"date\",\n", " y=[\"sales\", \"prediction\"],\n", " facet_row_spacing=0.04,\n", " facet_col=\"store_id\",\n", " facet_col_wrap=2,\n", " height=700,\n", " width=720,\n", ")\n", "fig.update_layout(\n", " legend=dict(orientation=\"h\", yanchor=\"top\", y=1.09, xanchor=\"center\", x=0.5),\n", " margin=dict(l=0, r=10, t=5, b=5),\n", " legend_title=\"\",\n", ")\n", "fig.update_traces(line=dict(width=1.7))\n", "fig.update_yaxes(matches=None, title=\"\")\n", "fig.update_xaxes(type=\"date\", range=[\"2015-11-01\", \"2016-05-22\"])" ] } ], "metadata": { "application/vnd.databricks.v1+notebook": { "dashboards": [ { "elements": [], "globalVars": {}, "guid": "ef82ffd4-2993-4b79-8327-f644b750f2dd", "layoutOption": { "grid": true, "stack": true }, "nuid": "a172d56a-d964-4505-ba66-5a7011220dbf", "origId": 1859120955398731, "title": "Untitled", "version": "DashboardViewV1", "width": 1024 } ], "language": "python", "notebookMetadata": { "pythonIndentUnit": 4 }, "notebookName": "ForecastFlowML Demo", "notebookOrigID": 2597536912577418, "widgets": {} }, "kernelspec": { "display_name": "spark", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.11.3" } }, "nbformat": 4, "nbformat_minor": 0 }