Databricksにおけるサプライチェーン配送の最適化

こちらのソリューションアクセラレータをウォークスルーします。

https://www.databricks.com/solutions/accelerators/supply-chain-distribution-optimization

Unity Catalogに対応するために一部修正を加えています。

# 外部ロケーションの作成

こちらのソリューションアクセラレータでは、テーブルを外部ロケーションに保存しています。外部ロケーションがない場合には事前に設定してください。

https://docs.databricks.com/ja/connect/unity-catalog/external-locations.html

# _resources/00-setup

カタログ名を指定しています。

```py
import os
import re 
import mlflow
db_prefix = "supply_chain_optimization"
catalogName = "takaakiyayoi_catalog"
```

外部ロケーションのパスを変更し、カタログを指定しています。また、Unity Catalog配下にストレージパスを指定したテーブルを作成する際には`MANAGED LOCATION`句が必要となります。

```py
# Get dbName and cloud_storage_path, reset and create database
current_user = dbutils.notebook.entry_point.getDbutils().notebook().getContext().tags().apply('user')
if current_user.rfind('@') > 0:
  current_user_no_at = current_user[:current_user.rfind('@')]
else:
  current_user_no_at = current_user
current_user_no_at = re.sub(r'\W+', '_', current_user_no_at)

dbName = db_prefix+"_"+current_user_no_at
#cloud_storage_path = f"/Users/{current_user}/field_demos/{db_prefix}"
cloud_storage_path = f"s3://taka-external-location-bucket/external-tables/{db_prefix}"

reset_all = dbutils.widgets.get("reset_all_data") == "true"

spark.sql(f"""USE CATALOG {catalogName}""")

if reset_all:
  spark.sql(f"DROP DATABASE IF EXISTS {dbName} CASCADE")
  dbutils.fs.rm(cloud_storage_path, True)

spark.sql(f"""create database if not exists {dbName} MANAGED LOCATION '{cloud_storage_path}' """)
spark.sql(f"""USE {dbName}""")
```

# _resources/01-data-generator

こちらは、`00-setup`から呼び出されますが、こちらでもカタログの指定が必要なので、ノートブックの冒頭でこちらを実行します。

```py
# Unity Catalog対応
catalogName = "takaakiyayoi_catalog"
spark.sql(f"""USE CATALOG {catalogName}""")
```

# イントロダクション

ノートブック`01_Introduction_And_Setup`を実行します。

**シチュエーション**:
モデリングする対象は電源ツールの製造業者のものです。この製造業者は3つのプラントを有し、5つの配送センターに30の商品SKUを配送しています。それぞれの配送センターには50から60のハードウェア店舗に割り当てられています。これらのパラメータの全ては、コードのパターンをスケールできるように取り扱われます(後ほどみていきます)。それぞれの店舗では、それぞれの商品に対する需要の時系列データがあります。

**提供物**:
- それぞれのハードウェア店舗におけるそれぞれの商品に対する需要の時系列データ
- ハードウェア店舗に対してそれぞれの配送センターを個別に割り当てるマッピングテーブル。あるハードウェア店舗が別の配送センターから商品を取得することはあり得ますが、ここでは簡素化しています。
- それぞれの製造プラントからそれぞれの配送センターに商品を配送する際のコストを割り当てるテーブル
- それぞれの配送センターにそれぞれのプラントから製造、配送できる商品の最大量を格納するテーブル

**2ステップで進めます**:
- *需要予測*: まず、一週間先の需要予測を行います。集計処理によって、それぞれの配送センターの需要を導出します: 
  - それぞれの店舗における個々の商品に対する需要予測で、一週間先の予測を生成します
  - 配送センターに対しては、それぞれの商品に対する翌週の需要を導出します
- *配送コストの最小化*: 製造およびプラントから配送センターへの配送に関するコストと制約のテーブルから、コストが最適化された輸送を導き出します。

![](https://raw.githubusercontent.com/databricks-industry-solutions/supply-chain-optimization/main/pictures/Plant_DistributionCenter_Store_Prouct_2.png)

```py
%run ./_resources/00-setup $reset_all_data=true
```
![Screenshot 2024-06-10 at 13.54.06.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/e44ed7e6-2a41-59f0-6a95-50846c22be2f.png)

上のコマンドが終了するといくつかのテーブルが作成されます。
![Screenshot 2024-06-10 at 13.55.00.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/cf175fcc-5702-00b2-38f5-740e33a1316f.png)

# 高精細な需要予測

ノートブック`02_Fine_Grained_Demand_Forecasting`を実行します。

*前提条件: このノートブックを実行する前に 01_Introduction_And_Setup を実行してください。*

このノートブックでは、それぞれの店舗と商品に対する翌週の需要を推定するために1週間先の予測を実行します。そして、それぞれの商品に対する配送センターレベルで集計を行います。

このノートブックにおけるハイライト:
- 適切な時系列モデルを特定するために、Databricksのコラボレーティブかつインタラクティブなノートブック環境の活用
- シングルノードのデータサイエンスコードを用いて複数ノードに分散させるためにPandas UDF(ユーザー定義関数)の活用

```py
%run ./_resources/00-setup $reset_all_data=false
```

```py
print(cloud_storage_path)
print(dbName)
```
```
s3://taka-external-location-bucket/external-tables/supply_chain_optimization
supply_chain_optimization_takaaki_yayoi
```

```py
import os
import datetime as dt
import numpy as np
import pandas as pd

from statsmodels.tsa.api import ExponentialSmoothing

import pyspark.sql.functions as f
from pyspark.sql.types import *
```

```py
demand_df = spark.read.table(f"{dbName}.part_level_demand")
demand_df = demand_df.cache() # このサンプルノートブックのためのもの
```

```py
display(demand_df)
```

日次での店舗ごと、商品ごとの需要データが確認できます。
![Screenshot 2024-06-10 at 13.57.06.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/fe119416-a89a-1e3e-1354-515a4658a17c.png)

## Holt’s Wintersの季節性手法を通じたワンステップ先の予測

Holt-Wintersの手法は、三重指数平滑法をベースとしており、トレンドと季節性の両方を考慮することができます。

[Exponential Smoothingの概要とアルゴリズム及び実装例について \| Deus Ex Machina](https://deus-ex-machina-ism.com/?p=57004#:~:text=%E4%B8%89%E9%87%8D%E6%8C%87%E6%95%B0%E5%B9%B3%E6%BB%91%E6%B3%95%EF%BC%88Triple,%E3%83%87%E3%83%BC%E3%82%BF%E3%81%AB%E9%81%A9%E3%81%97%E3%81%A6%E3%81%84%E3%82%8B%E3%80%82)

```py
def one_step_ahead_forecast(pdf: pd.DataFrame) -> pd.DataFrame:

    # 予測時系列の準備
    series_df = pd.Series(pdf['demand'].values, index=pdf['date'])
    series_df = series_df.asfreq(freq='W-MON')

    # ワンステップ先の予測
    fit1 = ExponentialSmoothing(
        series_df,
        trend="add",
        seasonal="add",
        use_boxcox=False,
        initialization_method="estimated",
    ).fit(method="ls")
    fcast1 = fit1.forecast(1).rename("Additive trend and additive seasonal")

    # 結果の取得
    df = pd.DataFrame(data = 
                      {
                         'product': pdf['product'].iloc[0], 
                         'store': pdf['store'].iloc[0], 
                         'date' : pd.to_datetime(series_df.index.values[-1]) + dt.timedelta(days=7), 
                         'demand' : int(abs(fcast1.iloc[-1]))
                      }, 
                          index = [0]
                     )
    return df
```

```py
fc_schema = StructType(
  [
    StructField('product', StringType()),
    StructField('store', StringType()),
    StructField('date', DateType()),
    StructField('demand', FloatType())
  ]
)
```

```py
spark.conf.set("spark.databricks.optimizer.adaptive.enabled", "false")
n_tasks = demand_df.select("product", "store").distinct().count()

forecast_df = (
  demand_df
  .repartition(n_tasks, "product", "store")
  .groupBy("product", "store")
  .applyInPandas(one_step_ahead_forecast, schema=fc_schema)
)

display(forecast_df)
```

一週間先の需要が予測できました(こちらを実行しているのは6/10です)。
![Screenshot 2024-06-10 at 13.58.44.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/72201b3b-820c-a050-176e-a8924cdb703f.png)

## 配送センターレベルでの予測の集計

```py
distribution_center_to_store_mapping_table = spark.read.table(f"{dbName}.distribution_center_to_store_mapping_table")
```

店舗と配送センターのマッピングを確認します。
```py
display(distribution_center_to_store_mapping_table)
```
![Screenshot 2024-06-10 at 13.59.54.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/1e8c2cdf-381f-add5-feb5-fc83a5bd1920.png)

予測結果とマッピングテーブルを結合します。
```py
distribution_center_demand = (
  forecast_df.
  join(distribution_center_to_store_mapping_table, [ "store" ] , "left").
  groupBy("distribution_center", "product").
  agg(f.sum("demand").alias("demand"))
)     
```

```py
display(distribution_center_demand)
```
![Screenshot 2024-06-10 at 14.01.00.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/d1ebc2d6-651c-4ff8-d593-85cec89ed9c5.png)

## Deltaとして保存

```py
distribution_center_demand_df_delta_path = os.path.join(cloud_storage_path, 'distribution_center_demand_df_delta')
```

```py
# データの書き込み 
distribution_center_demand.write \
.mode("overwrite") \
.format("delta") \
.save(distribution_center_demand_df_delta_path)
```

```py
spark.sql(f"DROP TABLE IF EXISTS {dbName}.distribution_center_demand")
spark.sql(f"CREATE TABLE {dbName}.distribution_center_demand USING DELTA LOCATION '{distribution_center_demand_df_delta_path}'")
```
```py
display(spark.sql(f"SELECT * FROM {dbName}.distribution_center_demand"))
```
![Screenshot 2024-06-10 at 14.02.01.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/63a994e8-6b80-b3a4-974a-1e7b7af6c16f.png)

# 輸送の最適化

ノートブック`03_Optimize_Transportation`を実行していきます。

*前提条件: このノートブックを実行する前に 02_Fine_Grained_Demand_Forecasting を実行してください。*

このノートブックでは、プラントから配送センターに商品を配送する際の輸送コストを最適化するためにLPを解きます。さらに、数十万の商品にどのようにスケールするのかを説明します。

[線形計画法と Optimizer \| Anapedia](https://help.anaplan.com/ja/%E7%B7%9A%E5%BD%A2%E8%A8%88%E7%94%BB%E6%B3%95%E3%81%A8-Optimizer-6a6489be-23fc-45f2-8afb-c98fa714508e)

このノートブックのハイライト:
- 最適経路を特定するためにDatabricksのコラボレーティブかつインタラクティブなノートブック環境の活用
- シングルノードのデータサイエンスコードを用いて複数ノードに分散させるためにPandas UDF(ユーザー定義関数)の活用

より正確には、それぞれの商品に対する以下の最適化問題を解くことになります。

*数学的なゴール:*

一連の配送センターに商品を配送する製造プラントを有しています。ここでのゴールは全体的な輸送コスト、すなわち以下の値を最小化することになります: 

> cost_of_plant_1_to_distribution_center_1 * quantity_shipped_of_plant_1_to_distribution_center_1
\+ … \+
cost_of_plant_1_to_distribution_center_n * quantity_shipped_of_plant_n_to_distribution_center_m 

*数学的な制約:*
- 配送数はゼロあるいは正の値であること
- ある製造プラントから出荷される商品の合計数は、最大供給量を上回ってはならない
- それぞれの配送センターに配送される商品の合計は、少なくとも予測された需要を満たさなくてはならない

```py
# LP問題を解くためにpulpライブラリを使います
# このノートブックの開発にはこちらのドキュメントを使いました
# https://coin-or.github.io/pulp/CaseStudies/a_transportation_problem.html
%pip install pulp
```

```py
%run ./_resources/00-setup $reset_all_data=false
```

```py
import os
import datetime as dt
import re
import numpy as np
import pandas as pd

import pulp

import pyspark.sql.functions as f
from pyspark.sql.types import *
```

## LPの定義および解決

```py
# それぞれの配送センターの需要、1行が1商品
distribution_center_demand = spark.read.table(f"{dbName}.distribution_center_demand")
distribution_center_demand = distribution_center_demand.groupBy("Product").pivot("distribution_center").agg(f.first("demand").alias("demand"))
for name in distribution_center_demand.schema.names:
  distribution_center_demand = distribution_center_demand.withColumnRenamed(name, name.replace("Distribution_Center", "Demand_Distribution_Center"))
distribution_center_demand = distribution_center_demand.withColumnRenamed("Product", "Product".replace("Product", "product")).sort("product")
display(distribution_center_demand)
```
![Screenshot 2024-06-10 at 14.04.31.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/fe152e0d-c786-48cd-ae9f-5935b7463d09.png)

```py
# プラントの供給、1行が1商品
plant_supply = spark.read.table(f"{dbName}.supply_table")
for name in plant_supply.schema.names:
  plant_supply = plant_supply.withColumnRenamed(name, name.replace("plant", "Supply_Plant"))
plant_supply = plant_supply.sort("product")
display(plant_supply)
```
![Screenshot 2024-06-10 at 14.04.57.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/28cb0e11-5c35-e64a-6db1-3768d8ab6c78.png)

```py
# 輸送コストテーブル、1行が商品、プラントごと
transport_cost_table = spark.read.table(f"{dbName}.transport_cost_table")
for name in transport_cost_table.schema.names:
  transport_cost_table = transport_cost_table.withColumnRenamed(name, name.replace("Distribution_Center", "Cost_Distribution_Center"))
display(transport_cost_table)
```
![Screenshot 2024-06-10 at 14.05.23.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/deab6ffc-0bbc-d4a5-ec6d-6d84d5e50796.png)

```py
# 処理を行うすべての情報を持つテーブルの作成。このテーブルの1行で、製品とプラントに対応し列方向で表現
# - プラント(行方向)から配送センター(列方向)への輸送コスト
# - それぞれのプラント(列方向)からのそれぞれの商品(行方向)を供給する際の制約
# - それぞれの配送センター(列方向)の需要を満たすそれぞれの商品(行方向)の需要の制約
lp_table_all_info = (transport_cost_table.
                     join(plant_supply, ["product"], how="inner").
                     join(distribution_center_demand, ["product"], how="inner")
                    )
display(lp_table_all_info)
```
![Screenshot 2024-06-10 at 14.05.47.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/e5768274-c3c8-7f4e-6d43-ad3b3156f1d1.png)

```py
# 最終結果テーブルの出力スキーマの定義
res_schema = StructType(
  [
    StructField('product', StringType()),
    StructField('plant', StringType()),
    StructField('distribution_center', StringType()),
    StructField('qty_shipped', IntegerType())
  ]
)
```
```py
# ある商品に対するLPを解く関数の定義
def transport_optimization(pdf: pd.DataFrame) -> pd.DataFrame:

  # プラントのリスト、これはプラントに関係するその他のデータ構造の順番を定義します
  plants_lst = sorted(pdf["plant"].unique().tolist())

  # 配送センターのリスト、これは配送センターに関係するその他のデータ構造の順番を定義します
  p = re.compile('^Cost_(Distribution_Center_\d)$')
  distribution_centers_lst = sorted([ s[5:] for s in list(pdf.columns.values) if p.search(s) ])

  # 取りうるすべての経路を定義
  routes = [(p, d) for p in plants_lst for d in distribution_centers_lst]

  # LP変数を格納するディクショナリーの作成。ディクショナリーへの参照キーはプラント名であり、配送センター名です。そしてデータはRoute_Tupleとなります。
  # (例えば、["plant_1"]["distribution_center_1"]: Route_plant_1_distribution_center_1)。下限を0にし、上限をNone、integerとして定義します。
  vars = pulp.LpVariable.dicts("Route", (plants_lst, distribution_centers_lst), 0, None, pulp.LpInteger)

  # 他の検索テーブルのサブセットを作成
  ss_prod = pdf[ "product" ][0]

  # コスト、配送センターとプラントの順序が問題となります
  transport_cost_table_pdf = pdf.filter(regex="^Cost_Distribution_Center_\d+$|^plant$")
  transport_cost_table_pdf = (transport_cost_table_pdf.
                              rename(columns=lambda x: re.sub("^Cost_Distribution_Center","Distribution_Center",x)).
                              set_index("plant").
                              reindex(plants_lst, axis=0).
                              reindex(distribution_centers_lst, axis=1)
                             )
  costs = pulp.makeDict([plants_lst, distribution_centers_lst], transport_cost_table_pdf.values.tolist(), 0)

  # 供給、プラントの順序が問題となります
  plant_supply_pdf = (pdf.
                      filter(regex="^Supply_Plant_\d+$").
                      drop_duplicates().
                      rename(columns=lambda x: re.sub("^Supply_Plant","plant",x)).
                      reindex(plants_lst, axis=1)
                     )

  supply = plant_supply_pdf.to_dict("records")[0]

  # 需要、配送センターの順序が問題となります
  distribution_center_demand_pdf =  (pdf.
                      filter(regex="^Demand_Distribution_Center_\d+$").
                      drop_duplicates().
                      rename(columns=lambda x: re.sub("^Demand_Distribution_Center","Distribution_Center",x)).
                      reindex(distribution_centers_lst, axis=1)
                     )

  demand = distribution_center_demand_pdf.to_dict("records")[0]

  # 問題データを含む'prob'変数の作成
  prob = pulp.LpProblem("Product_Distribution_Problem", pulp.LpMinimize)

  # まず、'prob'に対する目的関数を追加します
  prob += (
      pulp.lpSum([vars[p][d] * costs[p][d] for (p, d) in routes]),
      "Sum_of_Transporting_Costs",
  )

  # 供給の制約を追加します
  for p in plants_lst:
      prob += (
          pulp.lpSum([vars[p][d] for d in distribution_centers_lst]) <= supply[p],
          f"Sum_of_Products_out_of_Plant_{p}",
      )

  # 需要の制約を追加します
  for d in distribution_centers_lst:
      prob += (
          pulp.lpSum([vars[p][d] for p in plants_lst]) >= demand[d],
          f"Sum_of_Products_into_Distibution_Center{d}",
      )

  # PuLPのソルバー選択を用いて問題が解かれます
  prob.solve()

  # 商品ごとの出力を書き出します
  if (pulp.LpStatus[prob.status] == "Optimal"):
    name_lst = [ ]
    value_lst = [ ]
    for v in prob.variables():
      name_lst.append(v.name) 
      value_lst.append(v.varValue)
      res = pd.DataFrame(data={'name': name_lst, 'qty_shipped': value_lst})
      res[ "qty_shipped" ] = res[ "qty_shipped" ].astype("int")
      res[ "plant" ] =  res[ "name" ].str.extract(r'(plant_\d+)')
      res[ "distribution_center" ] =  res[ "name" ].str.extract(r'(Distribution_Center_\d+)')
      res[ "product" ] = ss_prod
      res = res.drop("name", axis = 1)
      res = res[[ "product", "plant", "distribution_center", "qty_shipped"]]
  else:
      res = pd.DataFrame(data= {  "product" : [ ss_prod ] , "plant" : [ None ], "distribution_center" : [ None ], "qty_shipped" : [ None ]})
  return res
```

```py
# 関数のテスト
product_selection = "nail_1"
pdf = lp_table_all_info.filter(f.col("product")==product_selection).toPandas()
transport_optimization(pdf)
```
![Screenshot 2024-06-10 at 14.06.32.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/1e883dce-72fc-3703-9386-3da7626cc03a.png)

Spark UDFを用いてLPを解きます。

```py
spark.conf.set("spark.databricks.optimizer.adaptive.enabled", "false")
n_tasks = lp_table_all_info.select("product").distinct().count()

optimal_transport_df = (
  lp_table_all_info
  .repartition(n_tasks, "product")
  .groupBy("product")
  .applyInPandas(transport_optimization, schema=res_schema)
)
```

## Deltaへの保存

```py
shipment_recommendations_df_delta_path = os.path.join(cloud_storage_path, 'shipment_recommendations_df_delta')
```

```py
# データの書き出し 
optimal_transport_df.write \
.mode("overwrite") \
.format("delta") \
.save(shipment_recommendations_df_delta_path)
```

```py
spark.sql(f"DROP TABLE IF EXISTS {dbName}.shipment_recommendations")
spark.sql(f"CREATE TABLE {dbName}.shipment_recommendations USING DELTA LOCATION '{shipment_recommendations_df_delta_path}'")
```

```py
display(spark.sql(f"SELECT * FROM {dbName}.shipment_recommendations"))
```

それぞれのプラントでそれぞれの商品をいくつ配送センターに輸送すべきかの推奨値が表示されます。
![Screenshot 2024-06-10 at 14.07.58.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/47c1e861-a0e3-55b5-2ec1-13c67145bb12.png)

ちなみに、今回作成されたテーブル間のリネージ(依存関係)はこのようになっています。
![Screenshot 2024-06-10 at 14.09.46.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/796a3ef8-6908-6c3e-40da-81ba1a6bc13d.png)


### はじめてのDatabricks

[はじめてのDatabricks](https://qiita.com/taka_yayoi/items/8dc72d083edb879a5e5d)

### Databricks無料トライアル

[Databricks無料トライアル](https://databricks.com/jp/try-databricks)