Deep learning on Spark with DL4J

この記事は[Spark Advent Calendar 9日目](http://qiita.com/advent-calendar/2015/apache-spark)の記事として書きました。

Spark上でDeep Learningのアルゴリズムを走らせるにはいくつか方法があります。

* MLlibで実装されている[MultilayerPerceptronClassifier](https://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.ml.classification.MultilayerPerceptronClassifier)を使う
* [Sparkling Water](http://h2o.ai/product/sparkling-water/)を使う
* [deeplearning4j](http://deeplearning4j.org/)を使う

今回は３つ目のdeeplearning4jをSparkから利用する方法を紹介したいと思います。

# deeplearning4jとは

[Skymind](http://www.skymind.io/)が中心となって開発をしているJVM上で動くDeep Learningのフレームワークです。Deep Learningのアルゴリズムを実装したものは[Caffe](http://caffe.berkeleyvision.org/)や[Torch](http://torch.ch/), [Chainer](http://chainer.org/)などがありますが、これらはメインではLuaやC/C++で実装されています。deeplearning4jはJava, Scalaで実装されており元々HadoopやSparkといったすでにあるBig Dataのためのフレームワークとの親和性に重きをおいて作られています。

![image](http://deeplearning4j.org/img/schematic_overview.png)
[Architecture from deeplearning4j.org](http://deeplearning4j.org/)

本体であるdeeplearning4jの他にも

* ベクトル化のためのツールである[Canova](https://github.com/deeplearning4j/Canova)
* JVM上でndarrayライクなベクトルを扱える[nd4j](https://github.com/deeplearning4j/nd4j)
* Scalaラッパである[deeplearning4s](https://github.com/deeplearning4j/deeplearning4s), [nd4s](https://github.com/deeplearning4j/nd4s)

などもともに開発されています。

# Sparkで利用するには

Sparkから利用するためにはこのdeeplearning4jを[Spark pacakge](http://spark-packages.org/)化したdl4j-spark-mlというパッケージを使います。

* [dl4j-spark-ml](http://spark-packages.org/package/deeplearning4j/dl4j-spark-ml)

すでにSparkが利用できる環境が整って入れば以下のコマンドを叩くだけです。

```bash
$ $SPARK_HOME/bin/spark-shell \
    --packages deeplearning4j:dl4j-spark-ml:0.4-rc3.4
```

これでclasspathにdeeplearning4jのクラスたちが通って、Spark上で利用できます。とはいってもどうアルゴリズムを書いていいかわからないので[dl4j-spark-ml-examples](https://github.com/deeplearning4j/dl4j-spark-ml-examples)という名前でサンプルコードも作られています。

ここに含まれているIrisデータセットを利用したサンプルを見てみましょう。

# [ml.JavaIrisClassification](https://github.com/deeplearning4j/dl4j-spark-ml-examples/blob/master/src/main/java/org/deeplearning4j/ml/JavaIrisClassification.java)

このコードではSparkのDaraFrameを使ってIrisデータセットを読み込んでいます。deeplearning4jの方でそれ用のコネクタ(Relation)が容易されています。

```java
        String path = args.length == 1 ? args[0]
                : "file://" + System.getProperty("user.dir") + "/data/svmLight/iris_svmLight_0.txt";
        DataFrame data = jsql.read()
                .format(DefaultSource.class.getName())
                .load(path);
```

このDataFrameに対してML Pipelineを構築していきます。Spark ML/MLlibの資産を利用できるのは大変便利です。

```java
        // 訓練用のデータを抜き出す
        DataFrame trainingData = data.sample(false, 0.6, 11L);
        // テスト用のデータを抜き出す
        DataFrame testData = data.except(trainingData);

        // 特徴量をscalingさせる
        StandardScaler scaler = new StandardScaler()
                .setWithMean(true).setWithStd(true)
                .setInputCol("features")
                .setOutputCol("scaledFeatures");
  
        // deeplearning4jが用意したEstimatorを作成する
        NeuralNetworkClassification classification 
          = new NeuralNetworkClassification()
                .setFeaturesCol("scaledFeatures")
                .setConf(getConfiguration());

        // Pipelineを作る
        Pipeline pipeline = new Pipeline()
          .setStages(new PipelineStage[] { scaler, classification });
```

`NeuralNetworkClassification`というEstimatorを作るための設定は以下のようになります。

```java

new NeuralNetConfiguration.Builder()
  .seed(11L) // 初期重み行列を生成するためのseed
  .iterations(100) // backpropagtionを行うiterationの回数
  .learningRate(1e-3f) // 重み更新を行うときの学習率
  .optimizationAlgo(OptimizationAlgorithm.LINE_GRADIENT_DESCENT) // backpropagationの際の勾配の計算方法
  .momentum(0.9) // 勾配更新の際に利用される値
  .constrainGradientToUnitNorm(true)
  .useDropConnect(true) // 過学習などを避ける学習速度を早めるためにDrop outするかどうか
  .list(2) // 層の数。入力層（データそのもの）は含まなない
  // 第一層 制約付きボルツマンマシン　隠れ層と可視層のユニットのタイプを設定
  .layer(0, new RBM.Builder(RBM.HiddenUnit.RECTIFIED, RBM.VisibleUnit.GAUSSIAN)
              .nIn(4) // 入力の大きさ Irisのfeatureの数
              .nOut(3) // 出力層。今回はそのままカテゴリ数になっている
              .weightInit(WeightInit.XAVIER)
              .activation("relu") // 活性化関数のタイプ Rectified Linear Unit
              .lossFunction(LossFunctions.LossFunction.RMSE_XENT)
              .updater(Updater.ADAGRAD) // RBMの勾配更新はAdaGradを使う
              .k(1) // # Contrastive Divergenceのiterationの数。RBM特有パラメタ
              .dropOut(0.5) // Drop Outをこの層で行う割合
              .build()
  ) 
  // 第二層　出力層
  .layer(1, new OutputLayer.Builder(LossFunctions.LossFunction.MCXENT)
              .nIn(3) // 入力の数
              .nOut(3) // 出力の数
              .activation("softmax") // 活性化関数。Classificationなのでsoftmax
              .weightInit(WeightInit.XAVIER)
              .updater(Updater.ADAGRAD) // Full connection layerの更新の方法
              .dropOut(0.5) // Drop Outをこの層で行う割合
              .build()
  ) 
  .build();
```

Deep Learningは設定すべきHyper Parameterが多くtuningやモデルの設定が難しくなってしまうことが多いと言われています。まさにそれを体現したようなConfigurationの生成です。
何かよいモデルのプリセットなどがいくつかあれば十分というケースもあると思うのでそういった場合には使いづらいかもしれませんが、逆に考えるとこれだけのパラメタの自由度があたえられていますのでモデルを選択できる幅は広がりそうです。

このIris Datasetに対するexampleを動かしたければscriptが用意されていますのでそれをたたけば大丈夫です。

```bash
$ ./bin/run-example org.deeplearning4j.ml.JavaIrisClassification 
```

ただデータによっては（Mnistとか）必要な容量が環境に対して大きい場合があるので`run-example`に記載されている`--driver-memory`と`--executor-memory`を増やす必要があるかもしれません。