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-3 Baseline Model`と`10-4 One Hot Encoding`、`10-5 Log Scale`の内容となります。

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

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

ノートブックはこちら

https://github.com/taka-yayoi/SparkSample/blob/main/Chapter10/10-3%20Baseline%20Model.py

https://github.com/taka-yayoi/SparkSample/blob/main/Chapter10/10-4%20One%20Hot%20Encoding.py

https://github.com/taka-yayoi/SparkSample/blob/main/Chapter10/10-5%20Log%20Scale.py

# ベースラインモデルの構築

トレーニングデータセットの平均`price`を計算して、テストデータセットに対する予測カラムとして使用し、結果を評価します。

```py
from pyspark.sql.functions import avg, lit

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)

avgPrice = float(trainDF.select(avg("price")).first()[0])
predDF = testDF.withColumn("avgPrediction", lit(avgPrice))
```

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

regressionMeanEvaluator = RegressionEvaluator(predictionCol="avgPrediction", labelCol="price", metricName="rmse")

print(f"The RMSE for predicting the average price is: {regressionMeanEvaluator.evaluate(predDF):.2f}")
```
```
The RMSE for predicting the average price is: 240.71
```

# ワンホットエンコーディング

このノートブックでは、モデルの追加の特徴量を追加し、カテゴリー型の特徴量をどのように取り扱うのかを議論します。

```py
filePath = "/databricks-datasets/learning-spark-v2/sf-airbnb/sf-airbnb-clean.parquet"
airbnbDF = spark.read.parquet(filePath)
```

## Train/Testの分割

アップルツーアップルで比較できるように、前回のノートブックと同じシードを用いて同じように80/20の分割を行いましょう(クラスター構成を変更していない場合に限ります！)。

```py
trainDF, testDF = airbnbDF.randomSplit([.8, .2], seed=42)
```

## オプション1: StringIndexer, OneHotEncoder, VectorAssembler

ここでは、カテゴリー型の変数に対してワンホットエンコーディング(OHE)を行います。使用する最初のアプローチでは、StringIndexer, OneHotEncoder, VectorAssemblerを組み合わせます。

最初に、文字列ラベルのカラムを、ラベルインデックスのMLカラムにマッピングするために`StringIndexer`を使用する必要があります　[Python](https://spark.apache.org/docs/latest/api/python/pyspark.ml.html#pyspark.ml.feature.StringIndexer)/[Scala](https://spark.apache.org/docs/latest/api/scala/#org.apache.spark.ml.feature.StringIndexer)。

そして、StringIndexerのアウトプットに`OneHotEncoder`を適用します [Python](https://spark.apache.org/docs/latest/api/python/pyspark.ml.html#pyspark.ml.feature.OneHotEncoder)/[Scala](https://spark.apache.org/docs/latest/api/scala/#org.apache.spark.ml.feature.OneHotEncoder)。

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

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

stringIndexer = StringIndexer(inputCols=categoricalCols, 
                              outputCols=indexOutputCols, 
                              handleInvalid="skip")
oheEncoder = OneHotEncoder(inputCols=indexOutputCols, 
                           outputCols=oheOutputCols)
```

これで、数値特徴量とOHEされたカテゴリー型の特徴量を組み合わせることができます。

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

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

## オプション2: RFormula

StringIndexerとOneHotEncoderに手動でどのカラムがカテゴリー型かを指定するのではなく、RFormulaは自動でこれらのことを行います [Python](https://spark.apache.org/docs/latest/api/python/pyspark.ml.html#pyspark.ml.feature.RFormula)/[Scala](https://spark.apache.org/docs/latest/api/scala/#org.apache.spark.ml.feature.RFormula)。

RFormulaを用いることで、文字列型のカラムがある場合、カテゴリー型の特徴量としてそれらを取り扱い、文字列のインデックスの作成とワンホットエンコーディングを行います。それ以外の場合は何もしません。そして、ワンホットエンコーディングされた特徴量と数値特徴量を、`features`と呼ばれる単一のベクトルにまとめます。

```py
from pyspark.ml.feature import RFormula

rFormula = RFormula(formula="price ~ .", featuresCol="features", labelCol="price", handleInvalid="skip")
```

## 線形回帰

すべての特徴量に対応したので、線形回帰モデルを構築しましょう。

```py
from pyspark.ml.regression import LinearRegression

lr = LinearRegression(labelCol="price", featuresCol="features")
```

## パイプライン

これらすべてのステージをパイプラインにまとめます。`Pipeline`はすべてのトランスフォーマーとエスティメーターを整理するための手段です [Python](https://spark.apache.org/docs/latest/api/python/pyspark.ml.html#pyspark.ml.Pipeline)/[Scala](https://spark.apache.org/docs/latest/api/scala/#org.apache.spark.ml.Pipeline)。

オプション1(StringIndexer, OneHotEncoderEstimator, VectorAssembler)とオプション2(RFormula)で同じ結果になることを検証します。

```py
# オプション 1: StringIndexer + OHE + VectorAssembler
from pyspark.ml import Pipeline

stages = [stringIndexer, oheEncoder, vecAssembler, lr]
pipeline = Pipeline(stages=stages)

pipelineModel = pipeline.fit(trainDF)
predDF = pipelineModel.transform(testDF)
predDF.select("features", "price", "prediction").show(5)
```
```
+--------------------+-----+------------------+
|            features|price|        prediction|
+--------------------+-----+------------------+
|(98,[0,3,6,22,43,...| 85.0| 55.86406437022288|
|(98,[0,3,6,22,43,...| 45.0| 22.87558895398388|
|(98,[0,3,6,22,43,...| 70.0|27.382602888721067|
|(98,[0,3,6,12,42,...|128.0|-91.50712171182795|
|(98,[0,3,6,12,43,...|159.0| 94.66621817641771|
+--------------------+-----+------------------+
only showing top 5 rows
```

```py
# オプション 2: RFormula
from pyspark.ml import Pipeline

pipeline = Pipeline(stages = [rFormula, lr])

pipelineModel = pipeline.fit(trainDF)
predDF = pipelineModel.transform(testDF)
predDF.select("features", "price", "prediction").show(5)
```
```
+--------------------+-----+------------------+
|            features|price|        prediction|
+--------------------+-----+------------------+
|(98,[0,3,6,7,23,4...| 85.0| 55.40518338251741|
|(98,[0,3,6,7,23,4...| 45.0|22.558643930734434|
|(98,[0,3,6,7,23,4...| 70.0|27.035891181432817|
|(98,[0,3,6,7,13,4...|128.0|-91.29310091873367|
|(98,[0,3,6,7,13,4...|159.0| 94.66473870534765|
+--------------------+-----+------------------+
only showing top 5 rows
```

## モデルの評価: RMSE

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

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

rmse = round(regressionEvaluator.evaluate(predDF), 2)
print(f"RMSE is {rmse}")
```
```
RMSE is 220.69
```

## R2

R2はどうなっているでしょうか？

```py
r2 = round(regressionEvaluator.setMetricName("r2").evaluate(predDF), 2)
print(f"R2 is {r2}")
```
```
R2 is 0.16
```

```py
pipelinePath = "/tmp/sf-airbnb/lr-pipeline-model"
pipelineModel.write().overwrite().save(pipelinePath)
```

## モデルのロード

モデルでロードする際、ロードし直すモデルのタイプを知っている必要があります(線形回帰、あるいはロジスティック回帰モデル？)。

このため、常に汎用的なPipelineModelをロードできるように、お使いのトランスフォーマーとエスティメーターをパイプラインに組み込むことをお勧めします。

```py
from pyspark.ml import PipelineModel

savedPipelineModel = PipelineModel.load(pipelinePath)
```

## 分散環境

分散環境において線形回帰がどのように実装されているのか、ボトルネックが何かを学ぶことに興味があるのであれば、以下の講義スライドをチェックしてください:
* [distributed-linear-regression-1](https://files.training.databricks.com/static/docs/distributed-linear-regression-1.pdf)
* [distributed-linear-regression-2](https://files.training.databricks.com/static/docs/distributed-linear-regression-2.pdf)

# ログスケール

このラボでは、ラベルをログスケールに変換し、ログスケールで予測を行い、結果を評価するために指数関数を取るようにすることでモデルのパフォーマンスを改善します。

```py
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)
```
```py
display(trainDF)
```
![Screenshot 2024-03-27 at 18.00.19.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/5b72a1ff-6fff-834b-850f-a175af3757f6.png)

`price`のヒストグラムを表示してみます。
![Screenshot 2024-03-27 at 18.00.33.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/561c2840-df6d-ee98-b2ca-75958aa8e464.png)
![Screenshot 2024-03-27 at 18.00.08.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/5a7188b3-bf6e-7c5f-3eba-bdb4348bc759.png)

対数分布しているようですので、ログスケールに変換します。

```py
from pyspark.sql.functions import col, log
from pyspark.ml import Pipeline
from pyspark.ml.feature import RFormula
from pyspark.ml.regression import LinearRegression

logTrainDF = trainDF.withColumn("log_price", log(col("price")))
logTestDF = testDF.withColumn("log_price", log(col("price")))

rFormula = RFormula(formula="log_price ~ . - price", featuresCol="features", labelCol="log_price", handleInvalid="skip") 

lr = LinearRegression(labelCol="log_price", predictionCol="log_pred")
pipeline = Pipeline(stages = [rFormula, lr])
pipelineModel = pipeline.fit(logTrainDF)
predDF = pipelineModel.transform(logTestDF)
```

## 指数関数

RMSEを解釈するためには、予測結果をログスケールから戻す必要があります。

```py
from pyspark.ml.evaluation import RegressionEvaluator
from pyspark.sql.functions import col, exp

expDF = predDF.withColumn("prediction", exp(col("log_pred")))

regressionEvaluator = RegressionEvaluator(labelCol="price", predictionCol="prediction")
rmse = regressionEvaluator.setMetricName("rmse").evaluate(expDF)
r2 = regressionEvaluator.setMetricName("r2").evaluate(expDF)
print(f"RMSE is {rmse}")
print(f"R2 is {r2}")
```
```
RMSE is 208.41093081345025
R2 is 0.25039912467770375
```

先ほどはRMSEは`220.69`だったので、精度が改善されていることがわかります。

### はじめてのDatabricks

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

### Databricks無料トライアル

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