【Databricks】作成したモデルをリモデルしてみる

この記事は株式会社ナレッジコミュニケーションが運営する クラウドAI by ナレコム Advent Calendar 2021 の3日目にあたる記事になります。

こんにちは！
今回は前回作成したモデルをリモデルしていきたいと思います。

モデルというのは、時間がたち新たなデータが増えてくることで規則性などに変化が起こり、予測精度がだんだん落ちてしまいます。
それを解消するためにリモデルを行います。

今回は精度を上げるためにでデータの形式を変更するので、このデータに合わせてリモデルしていきたいと思います。
この記事は前回の続きなので、前回のものと合わせて実行してください。

精度向上のアプローチ

前回作成したモデルの精度を上げたいなと思いました。また、今回利用したデータは数10MBと小さいデータですが実際には大容量のデータを利用する機会があると思います。
それに対応するためにStarbucks がとったアプローチを参考に店舗ごと、アイテムごとに分散処理を行うようにリモデルしたいと思います。

データの作成

前回同様データを作成します。
多少の変更はありますがほとんど前回と同様になっています。

notebook

sql_statement = '''
  SELECT
    store,
    item,
    date as ds,
    sales as y
  FROM
    history
  '''

store_item_history = (
  spark
    .sql( sql_statement )
    .repartition(sc.defaultParallelism, ['store', 'item'])
  ).cache()

display(store_item_history)

予測結果のフォーマットも作成しておきましょう。

notebook

from pyspark.sql.types import *

result_schema =StructType([
  StructField('ds',DateType()),
  StructField('store',IntegerType()),
  StructField('item',IntegerType()),
  StructField('y',FloatType()),
  StructField('yhat',FloatType()),
  StructField('yhat_upper',FloatType()),
  StructField('yhat_lower',FloatType())
  ])
'''

## モデルの作成
こちらも前回同様モデルを作成します。
前回と異なる点としては欠損データを削除している点です。

また、予測結果の結合を行っています。
これは、店舗・アイテムごとでモデルの作成、予測を行うため分かれているデータを結合する必要があるためです。

```python:notebook
from pyspark.sql.functions import pandas_udf, PandasUDFType

@pandas_udf( result_schema, PandasUDFType.GROUPED_MAP )
def forecast_store_item( history_pd ):
  
  # #1と同じロジックでモデルを作成
  # --------------------------------------
  # 欠損値を落とす(サブセットのデータ数によっては欠損値保管する必要あり)
  history_pd = history_pd.dropna()
  
  # モデル作成
  model = Prophet(
    interval_width=0.95,
    growth='linear', # linear or logistic
    daily_seasonality=False,
    weekly_seasonality=True,
    yearly_seasonality=True,
    seasonality_mode='multiplicative'
    )
  
  # モデル学習
  model.fit( history_pd )
  # --------------------------------------
  
  # #1と同じロジックで予測
  # --------------------------------------
  future_pd = model.make_future_dataframe(
    periods=90, 
    freq='d', 
    include_history=True
    )
  forecast_pd = model.predict( future_pd )  
  # --------------------------------------
  
  # サブセットを結合
  # --------------------------------------
  # 予測から関連フィールドを取得
  f_pd = forecast_pd[ ['ds','yhat', 'yhat_upper', 'yhat_lower'] ].set_index('ds')
  
  # 履歴から関連するフィールドを取得
  h_pd = history_pd[['ds','store','item','y']].set_index('ds')
  
  # 履歴と予測を結合
  results_pd = f_pd.join( h_pd, how='left' )
  results_pd.reset_index(level=0, inplace=True)
  
  # データセットから店舗と品番を取得
  results_pd['store'] = history_pd['store'].iloc[0]
  results_pd['item'] = history_pd['item'].iloc[0]
  # --------------------------------------
  
  # データセットを返す
  return results_pd[ ['ds', 'store', 'item', 'y', 'yhat', 'yhat_upper', 'yhat_lower'] ]  
'''

実際に実行していきたいと思います。
groupByを利用することで分割して予測することができます。

```python:notebook
from pyspark.sql.functions import current_date

results = (
  store_item_history
    .groupBy('store', 'item') # 分割して予測したいカラムを定義
    .apply(forecast_store_item) # 先のセルで定義した関数を適用 (モデリングと予測)
    .withColumn('training_date', current_date() ) # 予測を実行した日付のカラムを追加
    )

results.createOrReplaceTempView('new_forecasts')
'''

以下のクエリを実行することでモデルの作成、予測が実行されます。

```sql:notebook
%sql
SELECT * FROM new_forecasts

予測結果はクエリできるようにテーブルとして保存しておこうと思います。

notebook

%sql
-- create forecast table
create table if not exists forecasts (
  date date,
  store integer,
  item integer,
  sales float,
  sales_predicted float,
  sales_predicted_upper float,
  sales_predicted_lower float,
  training_date date
  )
using delta
partitioned by (training_date);

-- load data to it
insert into forecasts
select 
  ds as date,
  store,
  item,
  y as sales,
  yhat as sales_predicted,
  yhat_upper as sales_predicted_upper,
  yhat_lower as sales_predicted_lower,
  training_date
from new_forecasts;

モデルの評価

モデルを評価していきます。
pandasUDFを利用することで、店舗・アイテムごとに評価指標を算出できるので、それぞれで算出し出力したいと思います。

notebook

import pandas as pd

# 評価指標のカラム定義
eval_schema =StructType([
  StructField('training_date', DateType()),
  StructField('store', IntegerType()),
  StructField('item', IntegerType()),
  StructField('mae', FloatType()),
  StructField('mse', FloatType()),
  StructField('rmse', FloatType())
  ])

# 評価指標の算出
@pandas_udf( eval_schema, PandasUDFType.GROUPED_MAP )
def evaluate_forecast( evaluation_pd ):
  
  # データセットのストアとアイテムを取得
  training_date = evaluation_pd['training_date'].iloc[0]
  store = evaluation_pd['store'].iloc[0]
  item = evaluation_pd['item'].iloc[0]
  
  # 評価指標を算出
  mae = mean_absolute_error( evaluation_pd['y'], evaluation_pd['yhat'] )
  mse = mean_squared_error( evaluation_pd['y'], evaluation_pd['yhat'] )
  rmse = sqrt( mse )
  
  # 結果を結合
  results = {'training_date':[training_date], 'store':[store], 'item':[item], 'mae':[mae], 'mse':[mse], 'rmse':[rmse]}
  return pd.DataFrame.from_dict( results )

# calculate metrics
results = (
  spark
    .table('new_forecasts')
    .filter('ds < \'2018-01-01\'') # 評価を履歴データ(正解ラベル)がある期間に制限
    .select('training_date', 'store', 'item', 'y', 'yhat')
    .groupBy('training_date', 'store', 'item')
    .apply(evaluate_forecast)
    )
results.createOrReplaceTempView('new_forecast_evals')

こちらもクエリできるようにテーブルに出力しておきましょう。

notebook

%sql

create table if not exists forecast_evals (
  store integer,
  item integer,
  mae float,
  mse float,
  rmse float,
  training_date date
  )
using delta
partitioned by (training_date);

insert into forecast_evals
select
  store,
  item,
  mae,
  mse,
  rmse,
  training_date
from new_forecast_evals;
'''

それでは実際にどのような評価だったのか確認してみたいと思います。
まずは店舗1～10の製品1に限定して予測結果を確認してみます。

```sql:notebook
%sql

SELECT
  store,
  date,
  sales_predicted,
  sales_predicted_upper,
  sales_predicted_lower
FROM forecasts a
WHERE item = 1 AND
      --store IN (1, 2, 3, 4, 5) AND
      training_date=current_date() AND
      date >= '2018-01-01'
ORDER BY store

次に製品１について、店舗ごとの精度指標を確認してみましょう。

notebook

%sql

SELECT
  store,
  mae,
  mse,
  rmse
FROM forecast_evals a
WHERE --training_date=current_date() AND
  item = 1
ORDER BY store

このような結果になっています。
ここで前回の結果を合わせてみてみましょう。

-----------------------------
MSE: 1040.242690489999
RMSE: 32.25279353001844
MAE: 25.966749442943733

このような結果になっています。
すべての結果を確認しているわけではありませんが、どの指標もより良い数値になっていることがわかりますね。
リモデルに成功したようです!

まとめ

今回は前回作ったモデルに対して分散処理を取り入れるためにリモデルを行ってみました。
notebook上でデータを確認、生成しつつモデルを作成できるので、とても作成しやすい印象があります。

今回の記事はHow Starbucks Forecasts Demand at Scale with Databricksを参考にしています。