Sparkで機械学習入門 「価格推定」 #1 Apache Spark(Java)でデータセットの読み込み

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

[^1]: Sparkの**spark.ml**にある勾配ブースティングツリーは、いま流行ってる**LightGBM**、**XGBoost**(treeモデルを選択した場合)などと**勾配ブースティング**という基本的な理論は同じですが、設計面（並行性など）や性能面（過学習防止の為のハイパーパラメーターチューニングなど）はだいぶ異なっており同一のものではありません。Spark開発チームでもこれらを[ウォッチ](https://issues.apache.org/jira/browse/SPARK-8547)していたり、XGBoost本家による[xgboost4j-spark](https://github.com/dmlc/xgboost/tree/master/jvm-packages)などの提供もあります。

# 環境
- Apache Spark 2.4.3
- Java 8～

# 対象
- 難しいことを考えずに機械学習をまずはやってみようと思っている人

# やりたいこと

以下のようなアクセサリーの価格リストがあります。

「一番下の行にある**シルバーのブレスレットの値段**はいくらでしょうか？」

この値を**Java**と**Apache Spark**を使った機械学習によって予測したいとおもいます。

いわゆる**価格推定**です。


<table>
 <tr><td>id</td><td>材質</td><td>形状</td><td>重さ(g)</td><td>ブランド</td><td>販売店</td><td>価格(円)</td></tr>
<tr><td >0</td><td>シルバー</td><td>ブレスレット</td><td >56</td><td>海外有名ブランド</td><td>百貨店</td><td >40864</td></tr>
<tr><td >1</td><td>ゴールド</td><td>指輪</td><td >48</td><td>国内有名ブランド</td><td>直営店</td><td >63055</td></tr>
<tr><td >2</td><td>ダイア</td><td>イアリング</td><td >37</td><td>国内有名ブランド</td><td>直営店</td><td >112159</td></tr>
<tr><td >3</td><td>ダイア</td><td>ネックレス</td><td >20</td><td>海外有名ブランド</td><td>直営店</td><td >216053</td></tr>
<tr><td >4</td><td>ダイア</td><td>ネックレス</td><td >33</td><td>海外有名ブランド</td><td>百貨店</td><td >219666</td></tr>
<tr><td >5</td><td>シルバー</td><td>ブローチ</td><td >55</td><td>国内有名ブランド</td><td>百貨店</td><td >16482</td></tr>
<tr><td >6</td><td>プラチナ</td><td>ブローチ</td><td >58</td><td>海外超有名ブランド</td><td>直営店</td><td >377919</td></tr>
<tr><td >7</td><td>ゴールド</td><td>イアリング</td><td >49</td><td>国内有名ブランド</td><td>直営店</td><td >60484</td></tr>
<tr><td >8</td><td>シルバー</td><td>ネックレス</td><td >59</td><td>ノーブランド</td><td>激安ショップ</td><td >6256</td></tr>
<tr><td >9</td><td>ゴールド</td><td>指輪</td><td >13</td><td>国内有名ブランド</td><td>百貨店</td><td >37514</td></tr>
 <tr><td>・<br>・<br>・<br></td><td></td><td></td><td></td><td></td><td></td><td></td></tr>
 <tr><td>x</td><td>シルバー</td><td>ブレスレット</td><td>56</td><td>海外超有名ブランド</td><td>直営店</td><td><font color=red><b>お値段いくら？</b></font></td></tr>
</table>


#Apache SparkをローカルPCで使う

## 動作環境

### **Apache Sparkはスタンドアローン動作で**

**Apache Spark**は**Hadoop**の仲間でビッグデータを簡単に扱うことのできる**分散処理フレームワーク**で、複数台のマシンをクラスター構成で使う分散環境で強みを発揮しますが、<font color=blue>1台のローカルPCでのスタンドアローン動作も簡単</font>です。

今回は、**Apache Spark**をスタンドアローン動作させ**spark.ml**という機械学習機能を使います。

## OS
OSは**Javaランタイム**さえ入っていればWindows[^2]でもLinuxでもMacでもOKです

## 開発言語はJava

**Apache Spark**自体はScalaで書かれており、Scalaをはじめ、Java、Python、Rから利用できるAPIが整備されています。

本稿では、**Java**を使います。

## ライブラリ依存を設定する

Javaのアプリに**Apache Spark**を組み込んでつかうために依存ライブラリを設定します。

POM.xml(またはbuild.gradle)を準備して、以下のように**Apache Spark**関連ライブラリをdependenciesに加えるだけです。**Jackson**も**Apache Spark**内で使われるため追加しておきます。

```xml:POM.xml(抜粋)
<dependency>
	<groupId>org.apache.spark</groupId>
	<artifactId>spark-core_2.12</artifactId>
	<version>2.4.3</version>
</dependency>
<dependency>
	<groupId>org.apache.spark</groupId>
	<artifactId>spark-mllib_2.12</artifactId>
	<version>2.4.3</version>
</dependency>
<dependency>
	<groupId>com.fasterxml.jackson.module</groupId>
	<artifactId>jackson-module-scala_2.12</artifactId>
	<version>2.9.9</version>
</dependency>

<dependency>
	<groupId>com.fasterxml.jackson.core</groupId>
	<artifactId>jackson-databind</artifactId>
	<version>2.9.9</version>
</dependency>
```

## データセットを準備する

次に、学習に使うデータセットを準備しましょう。

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

今回は、ダイアモンドやプラチナなどの原材料でつくったネックレス・指輪などのアクセサリーの価格を機械学習で予測してみたいとおもいます。

機械学習に使うのは以下のような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
・・・・
```

このCSV形式データは[ここ](https://raw.githubusercontent.com/riversun/spark-gradient-boosting-tree-regression-example/master/dataset/gem_price_ja.csv)にのせましたが、**id、material(材質)、shape(形状)、weight(重さ)、brand(ブランド)、shop(販売店)、price(価格)**の順番で並んでおり、全部で**500件**あります。

つまり、変数としては**6種類**(idを除く**material,shape,weight,brand,shop,price**の6種類)あり、それが**500レコード**ということになります。

## Sparkでデータを読み込んでみる

ワークディレクトリの直下に**dataset**というディレクトリを作り学習用のデータ **gem_price_ja.csv**をそこに置きます。

この**CSV**ファイルを読み込んでSparkが扱えるようにするコードは以下の通り。

```java

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

public class GBTRegressionStep1 {

  public static void main(String[] args) {

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

    spark.sparkContext().setLogLevel("OFF");// (2)

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

    dataset.show();// (7)

    dataset.printSchema();// (8)

  }

}
```

### コード解説

**(1)** <i><b>.master("local[<span>\*</span>]")</b></i>・・・Sparkをローカルモードで動作させる。<b>「<span>\*</span>」</b>はワーカースレッドをCPUの論理コア数だけ割り当てる。***.master("local")***としたらワーカースレッドの数は**1**固定。***.master("local[2]")***としたらワーカースレッドの数は**2**となる。


**(2)**・・・ログをオフにする。

または**log4j**のほうで直接設定するには以下のようにします。

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

**(3)** ***.format("csv")*** ・・・データファイルをCSV形式として読み込む

**(4)** ***.option("header", "true")***・・・これをtrueにすると、CSVファイルの先頭の行をカラム名として使う
SparkではRDBMSのように、学習につかうためのデータを表形式のデータとして扱っていきます。学習データをCSV形式で準備する場合は、以下のように先頭の行にある定義をカラム名にマッピングできるので便利です。

```csv:CSVファイルの先頭の行
id,material,shape,weight,brand,shop,price
```

**(5)** ***.option("inferSchema", "true")***・・・入力データのスキーマ(schema)を推定する
これをtrueにすると、CSVファイルを読み込むときに１つ１つのデータの**型**をSparkが自動的に推定してセットしてくれます。便利ですね。
たとえば、以下の行ならシンプルなので、各データの**型**は**integer,string,string,integer,string,string,integer**のように推定されます。もし推定が間違えていたら、自分でスキーマを厳密に定義することも可能です。

**(6)** ***.load("dataset/gem_price_ja.csv");***・・・データファイルを読み込む

```csv:このようにシンプルなデータならinferSchemaで何とかなる
0,シルバー,ブレスレット,56,海外有名ブランド,百貨店,40864
```

**(7)** ***dataset.show();***・・・読み込んだデータセットを表示する

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

```
+---+--------+------------+------+------------------+------------+------+
| 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|
|  9|ゴールド|        指輪|    13|  国内有名ブランド|      百貨店| 37514|
| 10|プラチナ|  ネックレス|    23|  国内有名ブランド|激安ショップ| 48454|
| 11|  ダイア|  イアリング|    28|  海外有名ブランド|      直営店|233614|
| 12|シルバー|  ネックレス|    54|  国内有名ブランド|激安ショップ| 12235|
| 13|プラチナ|    ブローチ|    28|      ノーブランド|      百貨店| 34285|
| 14|シルバー|  ネックレス|    49|      ノーブランド|激安ショップ|  5970|
| 15|プラチナ|ブレスレット|    40|  国内有名ブランド|      百貨店| 82960|
| 16|シルバー|  ネックレス|    21|  海外有名ブランド|      百貨店| 28852|
| 17|ゴールド|        指輪|    11|  国内有名ブランド|      百貨店| 34980|
| 18|プラチナ|ブレスレット|    44|海外超有名ブランド|      百貨店|340849|
| 19|シルバー|        指輪|    11|海外超有名ブランド|      直営店| 47053|
+---+--------+------------+------+------------------+------------+------+
only showing top 20 rows

```

**(8)** ***dataset.printSchema();***・・・スキーマを表示する

さきほど***.option("inferSchema", "true")***しましたが、以下のようにSparkはちゃんと型が推定してくれました。

```
root
 |-- id: integer (nullable = true)
 |-- material: string (nullable = true)
 |-- shape: string (nullable = true)
 |-- weight: integer (nullable = true)
 |-- brand: string (nullable = true)
 |-- shop: string (nullable = true)
 |-- price: integer (nullable = true)
```

**[次回「#2 データの前処理（カテゴリ変数の取り扱い）」](https://qiita.com/riversun/items/070b8270b4381b6245d9)へと続く**

[^2]: SparkをWindows環境で実行するときに以下のようなエラーメッセージがでる場合は**winutils.exe**をダウンロードして適当なディレクトリに配置する。文末の補足にて説明

# 補足：WindowsでApache Sparkを実行する場合 

SparkをWindows環境で実行するときに以下のようなエラーメッセージがでる場合は**winutils.exe**をダウンロードして適当なディレクトリに配置する。

```
Could not locate executable null\bin\winutils.exe in the Hadoop binaries.
```

winutils.exeはUnixコマンドをWindowsでエミュレーションする為のユーティリティでHadoopが使う。

これを http://public-repo-1.hortonworks.com/hdp-win-alpha/winutils.exe からダウンロードしてきて、例えば、**c:/Temp/**以下に**c:/Temp/winutil/bin/winutil.exe**となるようにディレクトリをほって配置する。

そしてコードの先頭で以下のようにセットする。

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