Sparkで機械学習入門 「価格推定」 #2 データの前処理（カテゴリ変数の取り扱い）

# 概要
- **Java**と**Apache Spark**を使った機械学習入門編です
- 「価格推定」を題材に**Apache Spark**の使い方から実際の機械学習（学習、回帰）までステップアップ方式でハンズオンします
- 機械学習アルゴリズムには**教師あり学習**の**勾配ブースティングツリー**を使います
- ステップアップ方式に複数回に分けて投稿します
- 全ソースコードはこちら https://github.com/riversun/spark-gradient-boosting-tree-regression-example


# 環境
- Apache Spark 2.4.3
- Java 8～

# データの種類を意識して前処理をする

[前回](https://qiita.com/riversun/items/f1426118f66efddb327a)は、CSVデータからSparkの表形式データとして読み込むところまでやりました。

さて、ここで、今回学習に使うデータをもういちど眺めてみます。

データは[こちら](https://raw.githubusercontent.com/riversun/spark-gradient-boosting-tree-regression-example/master/dataset/gem_price_ja.csv)にあります。

こんなデータになっています。

```csv:gem_price_ja.csv
id,material,shape,weight,brand,shop,price
0,シルバー,ブレスレット,56,海外有名ブランド,百貨店,40864
1,ゴールド,指輪,48,国内有名ブランド,直営店,63055
2,ダイア,イアリング,37,国内有名ブランド,直営店,112159
3,ダイア,ネックレス,20,海外有名ブランド,直営店,216053
4,ダイア,ネックレス,33,海外有名ブランド,百貨店,219666
5,シルバー,ブローチ,55,国内有名ブランド,百貨店,16482
6,プラチナ,ブローチ,58,海外超有名ブランド,直営店,377919
7,ゴールド,イアリング,49,国内有名ブランド,直営店,60484
8,シルバー,ネックレス,59,ノーブランド,激安ショップ,6256
・・・・
```

## 質的データと量的データ

さて、変数名(materialとかshapeとか)ごとに、どんな値があるか見ていくと以下のように整理できました。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/170905/84d73bd6-67d8-850e-4495-f6aa869ae2e1.png)


こうしてみると、左側のデータ(**material,brand,shop,shape**)はそのカテゴリーを示す文字列の値になっています。このように、数値として直接測定できないものを**質的データ**と呼びます。

一方、右側のデータ(**weight,price**)は連続的な値をとる数値になっています。このように、数値として直接測定したり四則演算できるようなデータを**量的データ**と呼びます。

### カテゴリ変数(Categorical variable)と連続変数(Continuous variable)

前述のとおり、**質的データ**をとるような**material,brand,shop,shape**といった変数は、**カテゴリ変数(Categorical variable)**と呼びます。

また、**量的データ**、つまり連続する値(連続値)をとるような**weight,price**といった変数は、**連続変数(Categorical variable)**と呼びます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/170905/acffb1f0-f385-c704-569b-169b527e69cc.png)

### カテゴリ変数(Categorical variable)の取り扱い

機械学習は、結局のところ数値計算なので、カテゴリ変数がもつ質的データは何らかの数値に変換してから処理することになります。

ここでもう少しカテゴリ変数について見てみます。

カテゴリ変数の**material**と**shape**はアクセサリーの材質(material)と加工後の形態(**shape**)ですが、これらには意味的な違いがあります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/170905/4d60088f-866a-c8d7-ad09-20cfd03dd3e9.png)


**material**は***ダイア、プラチナ、ゴールド、シルバー***ですが、わたしたちは経験的にダイアはシルバーよりも高価であることを知っていますので、**material**がもつデータには序列がありそうです。

一方の**shape**は***指輪、ネックレス、イアリング、ブローチ、ブレスレット***ですが、指輪よりもブレスレットのほうが上等みたいな序列は存在せず、どの選択肢も対等な関係にあるように思えます。

### カテゴリ変数は順序変数(Ordinal variable)と名義変数(Nominal variable)に分けられる

このようにカテゴリー変数の値に序列がつくもの、大小を比較できるものを**順序変数(Ordinal variable)**（または順序尺度の変数）といいます。[^1]

[^1]: 今回の例だと、現在同じ重量のゴールドとプラチナだとゴールドのほうが高価なのでのどちらが上みたいなのがハッキリわからない感もありますが、わかりやすい例だと**上、中、下**とか**高、中、低**などがつく変数は順序変数と明確に定義することできるとおもいます。


反対にカテゴリー変数の値どうしに序列がつかないものは**名義変数(Nominal variable)**（または名義尺度の変数）と呼びます。

統計学の教科書のはじめの方に出てくる内容ですが、機械学習を進めていく上でも重要な知識なので覚えておきます。

今回のデータを整理してみると以下のような感じでしょうか。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/170905/8094f3c6-61e8-5dfc-2042-054ab644c7f1.png)

今回はいったんカテゴリ変数を順序変数と名義変数に明確に区別せずカテゴリ変数(Categorical variable)と連続変数(Continuous variable)というくくりで処理していきます。

## カテゴリ変数(Categorical variable)を数値化（インデックス化）する

上でも述べたように、機械学習するためにはカテゴリ変数を数値化する必要があります。

数値化の技法はいくつかありますが、今回は各変数にシンプルにインデックスをつけていきます。

↓こんなイメージ。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/170905/97c5428b-1bc7-5c15-1fb2-43eab66b4404.png)

## Sparkでカテゴリ変数を扱う

さて、カテゴリ変数を数値化ようコードを書いていきましょう。
まず以下のコードのようにします。

```java

import org.apache.spark.ml.feature.StringIndexer;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

public class GBTRegressionStep02 {

  public static void main(String[] args) {

    System.setProperty("hadoop.home.dir", "c:\\Temp\\winutil\\");

    org.apache.log4j.Logger.getLogger("org").setLevel(org.apache.log4j.Level.ERROR);
    org.apache.log4j.Logger.getLogger("akka").setLevel(org.apache.log4j.Level.ERROR);

    SparkSession spark = SparkSession
        .builder()
        .appName("GradientBoostingTreeGegression")
        .master("local[*]")
        .getOrCreate();

    Dataset<Row> dataset = spark
        .read()
        .format("csv")
        .option("header", "true")
        .option("inferSchema", "true")
        .load("dataset/gem_price_ja.csv");

    StringIndexer materialIndexer = new StringIndexer()// (1)
        .setInputCol("material")// (2)
        .setOutputCol("materialIndex");// (3)

    Dataset<Row> materialIndexAddedDataSet = materialIndexer.fit(dataset).transform(dataset);// (4)

    materialIndexAddedDataSet.show(10);// (5)

  }

}
```

### コード解説

**(1)**・・・カテゴリ変数を数値化(インデックス化)するためには***StringIndexer***を使います。ある変数を入力して、エンコード（変換）した結果を別の変数として出力する役割をもっています。ここでのエンコードは、単なる文字列を数値に変換する処理になります。

**(2)** ***.setInputCol("material")***・・・この***StringIndexer***に入力する変数名を指定します。CSVデータをSparkに読み込んだときに表形式の**Dataset**になっているので、**変数**は表の**カラム**と考えることができます。colは**カラム(column)**のcolですね。つまり、***StringIndexer**の入力側のカラム名は***"material"***あ

**(3)** ***.setOutputCol("materialIndex")***・・・こちらは、出力側のカラム名を指定する。

**(4)**・・・**fit()**メソッドでカテゴリ変数の数値化のための学習モデルをつくる。その結果得られた学習モデルに対してtransformメソッドでDatasetを与えると新しいDatasetを生成する。

>fitメソッドでカテゴリ変数の数値化のための学習モデルをつくる

ですが、
ここで**materialIndexer**という名前を付けた***StringIndexer***オブジェクトの**fit()**メソッドでは当たり前ですが実際には学習はしていません。単にカテゴリ変数として与えられた文字列を数値化する**仕掛け**[^2]を作っているだけです。

[^2]: ***StringIndexer***は**Estimator**というクラスを継承しています。**Estimator**は本来は**学習器**のためのクラスですが、**spark.ml**の特徴であるパイプライン処理をする際に、***StringIndexer***のような前処理系と学習器に同じインタフェースを備えておくことで　**前処理→前処理→学習**という流れをパイプラインとして実現するためにこのようなインタフェースになっています。

**transform**メソッドでは実際に入力となったDatasetにある**material**というカラム名から、**materialIndex**というインデックス化された新しい変数を作る処理が実行されます。

ここが一番意味不明だとおもいますが、あとで意義がわかるので、今は呪文でＯＫです。

**(5)**・・・**StringIndexer**を処理した後のDatasetを表示します。

コードを実行すると以下のようになります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/170905/9fb91367-3f45-2e43-37c5-51842a05b437.png)

**materialIndex**というカラムが右に追加されました！

このように、カテゴリ変数は何らかの形で数値化しないと機械学習で使えませんので、ほかのカテゴリ変数も数値化します。

ということで、おなじノリでほかのカテゴリ変数**shape,brand,shop**も同じように***StringIndexer**を使って数値化（インデックス化）してみます。

すると以下のようになります

```java
import org.apache.spark.ml.feature.StringIndexer;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

public class GBTRegressionStep02_part02 {

  public static void main(String[] args) {
    SparkSession spark = SparkSession
        .builder()
        .appName("GradientBoostingTreeGegression")
        .master("local[*]")
        .getOrCreate();

    Dataset<Row> dataset = spark
        .read()
        .format("csv")
        .option("header", "true")
        .option("inferSchema", "true")
        .load("dataset/gem_price_ja.csv");

    StringIndexer materialIndexer = new StringIndexer()// (1)
        .setInputCol("material")
        .setOutputCol("materialIndex");

    StringIndexer shapeIndexer = new StringIndexer()// (2)
        .setInputCol("shape")
        .setOutputCol("shapeIndex");

    StringIndexer brandIndexer = new StringIndexer()// (3)
        .setInputCol("brand")
        .setOutputCol("brandIndex");

    StringIndexer shopIndexer = new StringIndexer()// (4)
        .setInputCol("shop")
        .setOutputCol("shopIndex");

    Dataset<Row> dataset1 = materialIndexer.fit(dataset).transform(dataset);// (5)
    Dataset<Row> dataset2 = shapeIndexer.fit(dataset).transform(dataset1);// (6)
    Dataset<Row> dataset3 = brandIndexer.fit(dataset).transform(dataset2);// (7)
    Dataset<Row> dataset4 = shopIndexer.fit(dataset).transform(dataset3);// (8)

    dataset4.show(10);
  }
}
```

**(1)～(4)**は、それぞれの変数ごとに***StringIndexer***を準備しています。
**(5)～(8)**は、**materialIndexer→shapeIndexer→brandIndexer→shopIndexer**という順番で次々とインデックス化をしては、新しいDatasetを作る処理をしています。

これを実行すると以下のようになります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/170905/2250a5a8-cbd4-13d3-44ef-a9a637f6f597.png)

右に**materialIndex,shapeIndex,brandIndex,shopIndex**が追加されたことがわかります。

ところで、↓これは何とかならないのでしょうか。



```java
    Dataset<Row> dataset1 = materialIndexer.fit(dataset).transform(dataset);// (5)
    Dataset<Row> dataset2 = shapeIndexer.fit(dataset).transform(dataset1);// (6)
    Dataset<Row> dataset3 = brandIndexer.fit(dataset).transform(dataset2);// (7)
    Dataset<Row> dataset4 = shopIndexer.fit(dataset).transform(dataset3);// (8)
```

何とかなります。

**materialIndexer→shapeIndexer→brandIndexer→shopIndexer**のように、ある処理がおわったら、その処理を入力として次の処理をしてというのをもっとスマートに書くことができます。

それが次に見る**Pipeline**という仕組みです。

## データの前処理をパイプライン化する

↓処理は

```java:Before
    Dataset<Row> dataset1 = materialIndexer.fit(dataset).transform(dataset);
    Dataset<Row> dataset2 = shapeIndexer.fit(dataset).transform(dataset1);
    Dataset<Row> dataset3 = brandIndexer.fit(dataset).transform(dataset2);
    Dataset<Row> dataset4 = shopIndexer.fit(dataset).transform(dataset3);
```

***Pipeline***をつかって↓のように書き換えることができます。

```java:After
    Pipeline pipeline = new Pipeline()
        .setStages(new PipelineStage[] { materialIndexer, shapeIndexer, brandIndexer, shopIndexer });// (1)

    PipelineModel pipelineModel = pipeline.fit(dataset);// (2)

    Dataset<Row> indexedDataset = pipelineModel.transform(dataset);// (3)
```

**(1)**・・・このように***Pipeline***オブジェクトを作って、***setStages***で処理を実行したい順に***StringIndexer***を記述していきます。こうして並べるだけで、これら一連の処理を***Pipeline***として、ひとかたまりで実行することができます。

**(2)**・・・***pipeline#fit***で、pipelineModelを取得します。現段階でもまだ***StringIndexer***しか使っていないので学習用の処理は入れていませんが、学習用の処理が入ると***fit()***メソッドで得られる***PipelineModel***は指定したDatasetを元にした学習モデルを意味することになります。

**(3)**・・・***PipelineModel#transform***を実行すると一連の***StringIndexer***処理をまとめて実行し、結果として新しいDatasetを作ります。こちらも、Pipelineに学習用の処理を入れていた場合には、指定したDatasetに学習モデルを適用することを意味することになります。

以下にソースコードと実行結果を掲載します

```java

import org.apache.spark.ml.Pipeline;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.ml.PipelineStage;
import org.apache.spark.ml.feature.StringIndexer;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

public class GBTRegressionStep02_part03 {

  public static void main(String[] args) {

    System.setProperty("hadoop.home.dir", "c:\\Temp\\winutil\\");

    org.apache.log4j.Logger.getLogger("org").setLevel(org.apache.log4j.Level.ERROR);
    org.apache.log4j.Logger.getLogger("akka").setLevel(org.apache.log4j.Level.ERROR);

    SparkSession spark = SparkSession
        .builder()
        .appName("GradientBoostingTreeGegression")
        .master("local[*]")
        .getOrCreate();

    Dataset<Row> dataset = spark
        .read()
        .format("csv")
        .option("header", "true")
        .option("inferSchema", "true")
        .load("dataset/gem_price_ja.csv");

    StringIndexer materialIndexer = new StringIndexer()
        .setInputCol("material")
        .setOutputCol("materialIndex");

    StringIndexer shapeIndexer = new StringIndexer()
        .setInputCol("shape")
        .setOutputCol("shapeIndex");

    StringIndexer brandIndexer = new StringIndexer()
        .setInputCol("brand")
        .setOutputCol("brandIndex");

    StringIndexer shopIndexer = new StringIndexer()
        .setInputCol("shop")
        .setOutputCol("shopIndex");

    Pipeline pipeline = new Pipeline()
        .setStages(new PipelineStage[] { materialIndexer, shapeIndexer, brandIndexer, shopIndexer });// (1)

    PipelineModel pipelineModel = pipeline.fit(dataset);// (2)

    Dataset<Row> indexedDataset = pipelineModel.transform(dataset);// (3)
    indexedDataset.show(10);

  }

}
```

実行結果

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/170905/2250a5a8-cbd4-13d3-44ef-a9a637f6f597.png)

意味的に同じことをしているので、実行結果は同じです。

### インデクス化処理も、もう少しスマートに書く

さて、処理の実行の部分は***Pipeline***をつかってスッキリしましたが、以下の部分もちょっと冗長です。

```java
    StringIndexer materialIndexer = new StringIndexer()
        .setInputCol("material")
        .setOutputCol("materialIndex");

    StringIndexer shapeIndexer = new StringIndexer()
        .setInputCol("shape")
        .setOutputCol("shapeIndex");

    StringIndexer brandIndexer = new StringIndexer()
        .setInputCol("brand")
        .setOutputCol("brandIndex");

    StringIndexer shopIndexer = new StringIndexer()
        .setInputCol("shop")
        .setOutputCol("shopIndex");

    Pipeline pipeline = new Pipeline()
        .setStages(new PipelineStage[] { materialIndexer, shapeIndexer, brandIndexer, shopIndexer });/
```

ここはSparkというよりはJavaの書き方だけの問題ですが、少しスッキリさせてみます。

以下のように書き換えました。

```java
    List<String> categoricalColNames = Arrays.asList("material", "shape", "brand", "shop");// (1)

    List<StringIndexer> stringIndexers = categoricalColNames.stream()
        .map(col -> new StringIndexer()
            .setInputCol(col)
            .setOutputCol(col + "Index"))// (2)
        .collect(Collectors.toList());

    Pipeline pipeline = new Pipeline()
        .setStages(stringIndexers.toArray(new PipelineStage[0]));// (3)
```

**(1)**・・・カテゴリ変数名をListにします

**(2)**・・・***List#stream***をつかって***StringIndexer***を生成します。***setOutputCol***では出力側のカラム名を**カテゴリ変数名+"Index"**（つまり**materialIndex,shapeIndex,brandIndex,shopIndex**）となるようにしています。処理結果は***StringIndexer***の**List**になります。

**(3)**・・・Pipelineのステージに***StringIndexer***の配列になるようにセットします

ということで、今回のソースコードをまとめると以下のようになります。
だいぶスッキリしました。

```java

import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

import org.apache.spark.ml.Pipeline;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.ml.PipelineStage;
import org.apache.spark.ml.feature.StringIndexer;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

public class GBTRegressionStep02_part04 {

  public static void main(String[] args) {

    // System.setProperty("hadoop.home.dir", "c:\\Temp\\winutil\\");//for windows

    org.apache.log4j.Logger.getLogger("org").setLevel(org.apache.log4j.Level.ERROR);
    org.apache.log4j.Logger.getLogger("akka").setLevel(org.apache.log4j.Level.ERROR);

    SparkSession spark = SparkSession
        .builder()
        .appName("GradientBoostingTreeGegression")
        .master("local[*]")
        .getOrCreate();

    Dataset<Row> dataset = spark
        .read()
        .format("csv")
        .option("header", "true")
        .option("inferSchema", "true")
        .load("dataset/gem_price_ja.csv");

    List<String> categoricalColNames = Arrays.asList("material", "shape", "brand", "shop");

    List<StringIndexer> stringIndexers = categoricalColNames.stream()
        .map(col -> new StringIndexer()
            .setInputCol(col)
            .setOutputCol(col + "Index"))
        .collect(Collectors.toList());

    Pipeline pipeline = new Pipeline()
        .setStages(stringIndexers.toArray(new PipelineStage[0]));

    PipelineModel pipelineModel = pipeline.fit(dataset);

    Dataset<Row> indexedDataset = pipelineModel.transform(dataset);
    indexedDataset.show(10);

  }

}
```

これを実行するれば、先ほどと同じく、カテゴリ変数はすべてインデックス化された状態のdatasetを得ることができます。

## StringIndexerの数値化ポリシー

カテゴリ変数は順序変数(Ordinal variable)と名義変数に分けられる(Nominal variable)という話題を先にしましたが、Sparkの**StringIndexer**はどのようなポリシーでインデックスの数字を振っているのでしょうか。

**StringIndexer**はインデックスとして数字をつけて順序付けをしますが、その順序付けのルールはデフォルトだと、**出現頻度**となります。

**出現頻度**以外には、**アルファベット順**を指定することもできます。

以下に設定例を示します。

```java:出現頻度-降順(デフォルト)
    StringIndexer materialIndexer1 = new StringIndexer()
        .setStringOrderType("frequencyDesc")
        .setInputCol("material")
        .setOutputCol("materialIndex");
```

```java:アルファベット順-降順
    StringIndexer materialIndexer2 = new StringIndexer()
        .setStringOrderType("alphabetDesc")
        .setInputCol("material")
        .setOutputCol("materialIndex");
```

```java:アルファベット順-昇順
    StringIndexer materialIndexer3 = new StringIndexer()
        .setStringOrderType("alphabetAsc")
        .setInputCol("material")
        .setOutputCol("materialIndex");
```

今回は使いませんが、インデックスの序列に意味を持たせたいときには、変数名を調整して**setStringOrderType**で並び替えをする方法もあります。

# Apache Sparkの**spark.ml**のパイプライン(pipeline)処理まとめ

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/170905/5cc34aba-44af-39d6-35ef-69baedef091d.png)

**[次回「 #3 訓練データで学習させ、【価格推定エンジン】を作る」](https://qiita.com/riversun/items/8d2e74d41aa50b9707b6)へと続く**

次回は実際にデータを学習させて「価格推定エンジン」をつくります