Sparkによるハイパーパラメータチューニング

2024/4/12に翔泳社より**Apache Spark徹底入門**を出版します！

<a href="https://www.amazon.co.jp/dp/4798182281/"><img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/87910d7c-8a63-c198-8e82-216ea7a8ee07.jpeg" width=300></a>

書籍の[サンプルノートブック](https://github.com/databricks/LearningSparkV2/blob/master/notebooks/LearningSparkv2.dbc)をウォークスルーしていきます。`Python/Chapter10/10-7 Hyperparameter Tuning`となります。

翻訳ノートブックのリポジトリはこちら。

https://github.com/taka-yayoi/SparkSample/tree/main

ノートブックはこちら

https://github.com/taka-yayoi/SparkSample/blob/main/Chapter10/10-7%20Hyperparameter%20Tuning.py

ベストなハイパーパラメータを見つけ出すために、ランダムフォレストに対してハイパーパラメータチューニングを実施しましょう！

```py
from pyspark.ml.feature import StringIndexer, VectorAssembler

filePath = "/databricks-datasets/learning-spark-v2/sf-airbnb/sf-airbnb-clean.parquet"
airbnbDF = spark.read.parquet(filePath)
(trainDF, testDF) = airbnbDF.randomSplit([.8, .2], seed=42)

categoricalCols = [field for (field, dataType) in trainDF.dtypes if dataType == "string"]
indexOutputCols = [x + "Index" for x in categoricalCols]

stringIndexer = StringIndexer(inputCols=categoricalCols, outputCols=indexOutputCols, handleInvalid="skip")

numericCols = [field for (field, dataType) in trainDF.dtypes if ((dataType == "double") & (field != "price"))]
assemblerInputs = indexOutputCols + numericCols
vecAssembler = VectorAssembler(inputCols=assemblerInputs, outputCol="features")
```

# ランダムフォレスト

```py
from pyspark.ml.regression import RandomForestRegressor
from pyspark.ml import Pipeline

rf = RandomForestRegressor(labelCol="price", maxBins=40, seed=42)
pipeline = Pipeline(stages = [stringIndexer, vecAssembler, rf])
```

# グリッドサーチ

チューニング可能なハイパーパラメータは多数存在し、手動で設定するには長い時間を要します。

よりシステマティックなアプローチで最適なハイパーパラメータを見つけ出すために、Sparkの`ParamGridBuilder`を活用しましょう [Python](https://spark.apache.org/docs/latest/api/python/pyspark.ml.html#pyspark.ml.tuning.ParamGridBuilder)/[Scala](https://spark.apache.org/docs/latest/api/scala/#org.apache.spark.ml.tuning.ParamGridBuilder)。

テストするハイパーパラメータのグリッドを定義しましょう:
- maxDepth: 決定木の最大の深さ(`2, 4, 6`の値を使用)
- numTrees: 決定木の数(`10, 100`の値を使用)

```py
from pyspark.ml.tuning import ParamGridBuilder

paramGrid = (ParamGridBuilder()
            .addGrid(rf.maxDepth, [2, 4, 6])
            .addGrid(rf.numTrees, [10, 100])
            .build())
```

# 交差検証

最適なmaxDepthを特定するために、3フォールドの交差検証を活用します。

![crossValidation](https://files.training.databricks.com/images/301/CrossValidation.png)

3フォールドの交差検証によって、データの2/3でトレーニングを行い、(ホールドアウトされた)残りの1/3で評価を行います。このプロセスを3回繰り返すので、それぞれのフォールドは検証用セットとして動作する機会があります。そして、3ラウンドの結果を平均します。

以下を伝えるために、`CrossValidator`には`estimator`(パイプライン), `evaluator`, `estimatorParamMaps`を入力します:
- 使用するモデル
- モデルの評価方法
- モデルに設定するハイパーパラメータ

また、データを分割するフォールドの数を(3)に設定し、データが同じように分割されるようにシードも設定します [Python](https://spark.apache.org/docs/latest/api/python/pyspark.ml.html#pyspark.ml.tuning.CrossValidator)/[Scala](https://spark.apache.org/docs/latest/api/scala/#org.apache.spark.ml.tuning.CrossValidator)。

```py
from pyspark.ml.evaluation import RegressionEvaluator
from pyspark.ml.tuning import CrossValidator

evaluator = RegressionEvaluator(labelCol="price", 
                                predictionCol="prediction", 
                                metricName="rmse")

cv = CrossValidator(estimator=pipeline, 
                    evaluator=evaluator, 
                    estimatorParamMaps=paramGrid, 
                    numFolds=3, 
                    seed=42)
```

**問題**: この時点でいくつのモデルをトレーニングしていますか？

```py
cvModel = cv.fit(trainDF)
```
![Screenshot 2024-03-27 at 18.17.03.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/268c2481-52df-1029-2e5b-102981050a82.png)


# Parallelismパラメーター

うーん...実行に長い時間を要しています。これは、並列ではなく直列でモデルがトレーニングされているからです！

Spark 2.3では、[parallelism](https://spark.apache.org/docs/latest/api/python/pyspark.ml.html#pyspark.ml.tuning.CrossValidator.parallelism)パラメータが導入されました。ドキュメントでは、`並列アルゴリズムを実行する際のスレッド数 (>= 1)`と述べられています。

この値を4に設定し、トレーニングが早くなるかどうかを見てみましょう。

```py
cvModel = cv.setParallelism(4).fit(trainDF)
```
![Screenshot 2024-03-27 at 18.17.28.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/5f5cba9d-5f84-0fcb-fc28-5f76beb98060.png)


**問題**: うーん...依然として時間がかかっています。交差検証器の中にパイプラインを埋め込むべきか、パイプラインに交差検証器を埋め込むべきでしょうか？

パイプラインにエスティメーターやトランスフォーマーが含まれるかに依存します。StringIndexer(エスティメーター)のようなものがパイプラインにある場合、交差検証器にパイプライン全体を埋め込むと、毎回再フィットさせなくてはなりません。

```py
cv = CrossValidator(estimator=rf, 
                    evaluator=evaluator, 
                    estimatorParamMaps=paramGrid, 
                    numFolds=3, 
                    parallelism=4, 
                    seed=42)

pipeline = Pipeline(stages=[stringIndexer, vecAssembler, cv])

pipelineModel = pipeline.fit(trainDF)
```
![Screenshot 2024-03-27 at 18.17.53.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/3e2ffe3b-0aaa-17b6-66b0-48cc6d65f478.png)

ベストなハイパーパラメータの設定を持つモデルを見てみましょう。

```py
list(zip(cvModel.getEstimatorParamMaps(), cvModel.avgMetrics))
```
```
[({Param(parent='RandomForestRegressor_cee37cf57669', name='maxDepth', doc='Maximum depth of the tree. (>= 0) E.g., depth 0 means 1 leaf node; depth 1 means 1 internal node + 2 leaf nodes. Must be in range [0, 30].'): 2,
   Param(parent='RandomForestRegressor_cee37cf57669', name='numTrees', doc='Number of trees to train (>= 1).'): 10},
  291.1822640924783),
 ({Param(parent='RandomForestRegressor_cee37cf57669', name='maxDepth', doc='Maximum depth of the tree. (>= 0) E.g., depth 0 means 1 leaf node; depth 1 means 1 internal node + 2 leaf nodes. Must be in range [0, 30].'): 2,
   Param(parent='RandomForestRegressor_cee37cf57669', name='numTrees', doc='Number of trees to train (>= 1).'): 100},
  286.7714750274078),
 ({Param(parent='RandomForestRegressor_cee37cf57669', name='maxDepth', doc='Maximum depth of the tree. (>= 0) E.g., depth 0 means 1 leaf node; depth 1 means 1 internal node + 2 leaf nodes. Must be in range [0, 30].'): 4,
   Param(parent='RandomForestRegressor_cee37cf57669', name='numTrees', doc='Number of trees to train (>= 1).'): 10},
  287.6963245160818),
 ({Param(parent='RandomForestRegressor_cee37cf57669', name='maxDepth', doc='Maximum depth of the tree. (>= 0) E.g., depth 0 means 1 leaf node; depth 1 means 1 internal node + 2 leaf nodes. Must be in range [0, 30].'): 4,
   Param(parent='RandomForestRegressor_cee37cf57669', name='numTrees', doc='Number of trees to train (>= 1).'): 100},
  279.9927057236079),
 ({Param(parent='RandomForestRegressor_cee37cf57669', name='maxDepth', doc='Maximum depth of the tree. (>= 0) E.g., depth 0 means 1 leaf node; depth 1 means 1 internal node + 2 leaf nodes. Must be in range [0, 30].'): 6,
   Param(parent='RandomForestRegressor_cee37cf57669', name='numTrees', doc='Number of trees to train (>= 1).'): 10},
  294.34810870889305),
 ({Param(parent='RandomForestRegressor_cee37cf57669', name='maxDepth', doc='Maximum depth of the tree. (>= 0) E.g., depth 0 means 1 leaf node; depth 1 means 1 internal node + 2 leaf nodes. Must be in range [0, 30].'): 6,
   Param(parent='RandomForestRegressor_cee37cf57669', name='numTrees', doc='Number of trees to train (>= 1).'): 100},
  275.39862704729984)]
```

テストデータセットでどうなるのかを見てみましょう。

```py
predDF = pipelineModel.transform(testDF)

regressionEvaluator = RegressionEvaluator(predictionCol="prediction", labelCol="price", metricName="rmse")

rmse = regressionEvaluator.evaluate(predDF)
r2 = regressionEvaluator.setMetricName("r2").evaluate(predDF)
print(f"RMSE is {rmse}")
print(f"R2 is {r2}")
```
```
RMSE is 211.70370310223277
R2 is 0.2265254865671944
```

[前回のピュアな決定木の結果](https://qiita.com/taka_yayoi/items/790734663e639db956e9#%E3%83%86%E3%82%B9%E3%83%88%E3%82%BB%E3%83%83%E3%83%88%E3%81%AB%E3%83%A2%E3%83%87%E3%83%AB%E3%82%92%E9%81%A9%E7%94%A8)よりも精度が改善されました。

### はじめてのDatabricks

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

### Databricks無料トライアル

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