MLlib: メインガイド

MLlib: RDDベースAPIガイド

最適化 - RDDベースAPI

\[ \newcommand{\R}{\mathbb{R}} \newcommand{\E}{\mathbb{E}} \newcommand{\x}{\mathbf{x}} \newcommand{\y}{\mathbf{y}} \newcommand{\wv}{\mathbf{w}} \newcommand{\av}{\mathbf{\alpha}} \newcommand{\bv}{\mathbf{b}} \newcommand{\N}{\mathbb{N}} \newcommand{\id}{\mathbf{I}} \newcommand{\ind}{\mathbf{1}} \newcommand{\0}{\mathbf{0}} \newcommand{\unit}{\mathbf{e}} \newcommand{\one}{\mathbf{1}} \newcommand{\zero}{\mathbf{0}} \]

数学的説明

勾配降下法

形式 $\min_{\wv \in\R^d} \; f(\wv)$ の最適化問題を解く最も簡単な方法は 勾配降下法 です。このような一次最適化手法（勾配降下法とその確率的変種を含む）は、大規模で分散された計算に適しています。

勾配降下法は、現在の点における関数の最も急な下り坂の方向（現在のパラメータ値における関数の導関数（勾配）の負の値）に沿って反復的にステップを踏むことによって、関数の局所的最小値を見つけることを目指します。目的関数 $f$ がすべての引数で微分不可能であっても、凸である場合、劣勾配は勾配の自然な一般化であり、ステップ方向の役割を果たします。いずれにしても、$f$ の勾配または劣勾配を計算することはコストがかかります。これは、すべての損失項からの寄与を計算するために、完全なデータセット全体を一度走査する必要があるためです。

確率的勾配降下法 (SGD)

目的関数 $f$ が合計として表される最適化問題は、特に確率的勾配降下法 (SGD) を使用して解決するのに適しています。我々のケースでは、教師あり機械学習で一般的に使用される最適化定式化において、\begin{equation} f(\wv) := \lambda\, R(\wv) + \frac1n \sum_{i=1}^n L(\wv;\x_i,y_i) \label{eq:regPrimal} \ . \end{equation}これは特に自然です。なぜなら、損失は各データポイントからの個々の損失の平均として記述されるからです。

確率的劣勾配は、期待値で元の目的関数の真の劣勾配を得られるような、ベクトルのランダムな選択です。データポイント $i\in[1..n]$ を一様ランダムに1つ選択することにより、$\eqref{eq:regPrimal}$ の確率的劣勾配を $\wv$ に関して以下のように得られます。 \[ f'_{\wv,i} := L'_{\wv,i} + \lambda\, R'_\wv \ , \] ここで $L'_{\wv,i} \in \R^d$ は $i$ 番目のデータポイントによって決定される損失関数の部分の劣勾配、すなわち $L'_{\wv,i} \in \frac{\partial}{\partial \wv} L(\wv;\x_i,y_i)$ です。さらに、$R'_\wv$ は正則化項 $R(\wv)$ の劣勾配、すなわち $R'_\wv \in \frac{\partial}{\partial \wv} R(\wv)$ です。$R'_\wv$ という項は、どのランダムなデータポイントが選択されたかには依存しません。明らかに、$i\in[1..n]$ のランダムな選択に関する期待値において、$f'_{\wv,i}$ は元の目的関数 $f$ の劣勾配であり、$\E\left[f'_{\wv,i}\right] \in \frac{\partial}{\partial \wv} f(\wv)$ を意味します。

SGDを実行することは、負の確率的劣勾配 $f'_{\wv,i}$ の方向に歩くことに単純化されます。すなわち、\begin{equation}\label{eq:SGDupdate} \wv^{(t+1)} := \wv^{(t)} - \gamma \; f'_{\wv,i} \ . \end{equation}ステップサイズ。 パラメータ $\gamma$ はステップサイズです。デフォルトの実装では、これは反復カウンタの平方根に反比例するように選択されます。すなわち、$t$ 番目の反復では $\gamma := \frac{s}{\sqrt{t}}$ となり、入力パラメータは $s=$ stepSize です。SGD手法で最適なステップサイズを選択することは、実際にはしばしばデリケートであり、活発な研究トピックであることを注意してください。

勾配。 spark.mllib で実装されている機械学習手法の（劣）勾配の表は、分類と回帰のセクションで利用可能です。

近接更新。 ステップ方向における正則化項の劣勾配 $R'(\wv)$ を単に使用する代わりに、場合によっては近接演算子を使用することで改善された更新が得られます。L1正則化項の場合、近接演算子はソフト閾値処理によって与えられ、これは L1Updater に実装されています。

分散SGDの更新スキーム

GradientDescent のSGD実装では、データ例の単純な（分散）サンプリングを使用します。最適化問題 $\eqref{eq:regPrimal}$ の損失部分は $\frac1n \sum_{i=1}^n L(\wv;\x_i,y_i)$ であり、したがって $\frac1n \sum_{i=1}^n L'_{\wv,i}$ が真の（劣）勾配となることを思い出してください。これには完全なデータセットへのアクセスが必要となるため、miniBatchFraction パラメータは、代わりに完全なデータのどの割合を使用するかを指定します。このサブセットでの勾配の平均、すなわち \[ \frac1{|S|} \sum_{i\in S} L'_{\wv,i} \ , \] は確率勾配です。ここで $S$ はサイズ $|S|=$ miniBatchFraction $\cdot n$ のサンプリングされたサブセットです。

各反復において、分散データセット（RDD）のサンプリング、および各ワーカーマシンからの部分結果の合計の計算は、標準のSparkルーチンによって実行されます。

データポイントの割合 miniBatchFraction が 1 (デフォルト) に設定されている場合、各反復での結果ステップは正確な（劣）勾配降下法になります。この場合、ランダム性も使用されるステップ方向の分散もありません。極端な場合、miniBatchFraction が非常に小さく選択され、単一のデータポイントのみがサンプリングされる場合、すなわち $|S|=$ miniBatchFraction $\cdot n = 1$ の場合、アルゴリズムは標準のSGDと同等になります。その場合、ステップ方向はデータポイントの一様ランダムサンプリングに依存します。

限定記憶BFGS (L-BFGS)

L-BFGS は、準ニュートン法のファミリに属する最適化アルゴリズムであり、形式 $\min_{\wv \in\R^d} \; f(\wv)$ の最適化問題を解きます。L-BFGS法は、目的関数の2次偏導関数を評価してヘッセ行列を構築することなく、目的関数を局所的に二次関数として近似します。ヘッセ行列は過去の勾配評価によって近似されるため、ヘッセ行列を明示的に計算するニュートン法における垂直スケーラビリティの問題（トレーニング特徴量の数）はありません。その結果、L-BFGSはしばしば他の一次最適化手法と比較してより速い収束を達成します。

最適化手法の選択

線形手法は内部で最適化を使用し、spark.mllib の一部の線形手法はSGDとL-BFGSの両方をサポートします。異なる最適化手法は、目的関数のプロパティに応じて異なる収束保証を持つ可能性があり、ここでは文献を網羅することはできません。一般的に、L-BFGSが利用可能な場合は、SGDの代わりにそれを使用することをお勧めします。なぜなら、L-BFGSはより速く（より少ない反復で）収束する傾向があるからです。

MLlibでの実装

勾配降下法と確率的勾配降下法

確率的劣勾配降下法 (SGD) を含む勾配降下法は、MLlib に低レベルプリミティブとして含まれており、さまざまなMLアルゴリズムがその上に構築されています。例として、線形手法のセクションを参照してください。

SGD クラス GradientDescent は、以下のパラメータを設定します。

L-BFGS

L-BFGSは現在、MLlib では低レベルの最適化プリミティブにすぎません。線形回帰やロジスティック回帰などのさまざまなMLアルゴリズムでL-BFGSを使用したい場合は、LogisticRegressionWithSGD のようなトレーニングAPIを使用する代わりに、目的関数の勾配とupdaterを自分でoptimizerに渡す必要があります。以下の例を参照してください。これは次期リリースで対応される予定です。

L1Updater を使用したL1正則化は、L1Updaterのソフト閾値処理ロジックが勾配降下法のために設計されているため機能しません。開発者向けノートを参照してください。

L-BFGSメソッド LBFGS.runLBFGS は、以下のパラメータを持っています。

返される値は、2つの要素を含むタプルです。最初の要素は、各特徴量の重みを含む列行列であり、2番目の要素は、各反復で計算された損失を含む配列です。

以下は、L-BFGSオプティマイザを使用してL2正則化で二項ロジスティック回帰をトレーニングする例です。

APIの詳細については、LBFGS ScalaドキュメントおよびSquaredL2Updater Scalaドキュメントを参照してください。

import org.apache.spark.mllib.classification.LogisticRegressionModel
import org.apache.spark.mllib.evaluation.BinaryClassificationMetrics
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.optimization.{LBFGS, LogisticGradient, SquaredL2Updater}
import org.apache.spark.mllib.util.MLUtils

val data = MLUtils.loadLibSVMFile(sc, "data/mllib/sample_libsvm_data.txt")
val numFeatures = data.take(1)(0).features.size

// Split data into training (60%) and test (40%).
val splits = data.randomSplit(Array(0.6, 0.4), seed = 11L)

// Append 1 into the training data as intercept.
val training = splits(0).map(x => (x.label, MLUtils.appendBias(x.features))).cache()

val test = splits(1)

// Run training algorithm to build the model
val numCorrections = 10
val convergenceTol = 1e-4
val maxNumIterations = 20
val regParam = 0.1
val initialWeightsWithIntercept = Vectors.dense(new Array[Double](numFeatures + 1))

val (weightsWithIntercept, loss) = LBFGS.runLBFGS(
  training,
  new LogisticGradient(),
  new SquaredL2Updater(),
  numCorrections,
  convergenceTol,
  maxNumIterations,
  regParam,
  initialWeightsWithIntercept)

val model = new LogisticRegressionModel(
  Vectors.dense(weightsWithIntercept.toArray.slice(0, weightsWithIntercept.size - 1)),
  weightsWithIntercept(weightsWithIntercept.size - 1))

// Clear the default threshold.
model.clearThreshold()

// Compute raw scores on the test set.
val scoreAndLabels = test.map { point =>
  val score = model.predict(point.features)
  (score, point.label)
}

// Get evaluation metrics.
val metrics = new BinaryClassificationMetrics(scoreAndLabels)
val auROC = metrics.areaUnderROC()

println("Loss of each step in training process")
loss.foreach(println)
println(s"Area under ROC = $auROC")
完全なサンプルコードは、Sparkリポジトリの「examples/src/main/scala/org/apache/spark/examples/mllib/LBFGSExample.scala」にあります。

APIの詳細については、LBFGS JavaドキュメントおよびSquaredL2Updater Javaドキュメントを参照してください。

import java.util.Arrays;

import scala.Tuple2;

import org.apache.spark.api.java.*;
import org.apache.spark.mllib.classification.LogisticRegressionModel;
import org.apache.spark.mllib.evaluation.BinaryClassificationMetrics;
import org.apache.spark.mllib.linalg.Vector;
import org.apache.spark.mllib.linalg.Vectors;
import org.apache.spark.mllib.optimization.*;
import org.apache.spark.mllib.regression.LabeledPoint;
import org.apache.spark.mllib.util.MLUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;

String path = "data/mllib/sample_libsvm_data.txt";
JavaRDD<LabeledPoint> data = MLUtils.loadLibSVMFile(sc, path).toJavaRDD();
int numFeatures = data.take(1).get(0).features().size();

// Split initial RDD into two... [60% training data, 40% testing data].
JavaRDD<LabeledPoint> trainingInit = data.sample(false, 0.6, 11L);
JavaRDD<LabeledPoint> test = data.subtract(trainingInit);

// Append 1 into the training data as intercept.
JavaPairRDD<Object, Vector> training = data.mapToPair(p ->
  new Tuple2<>(p.label(), MLUtils.appendBias(p.features())));
training.cache();

// Run training algorithm to build the model.
int numCorrections = 10;
double convergenceTol = 1e-4;
int maxNumIterations = 20;
double regParam = 0.1;
Vector initialWeightsWithIntercept = Vectors.dense(new double[numFeatures + 1]);

Tuple2<Vector, double[]> result = LBFGS.runLBFGS(
  training.rdd(),
  new LogisticGradient(),
  new SquaredL2Updater(),
  numCorrections,
  convergenceTol,
  maxNumIterations,
  regParam,
  initialWeightsWithIntercept);
Vector weightsWithIntercept = result._1();
double[] loss = result._2();

LogisticRegressionModel model = new LogisticRegressionModel(
  Vectors.dense(Arrays.copyOf(weightsWithIntercept.toArray(), weightsWithIntercept.size() - 1)),
  (weightsWithIntercept.toArray())[weightsWithIntercept.size() - 1]);

// Clear the default threshold.
model.clearThreshold();

// Compute raw scores on the test set.
JavaPairRDD<Object, Object> scoreAndLabels = test.mapToPair(p ->
  new Tuple2<>(model.predict(p.features()), p.label()));

// Get evaluation metrics.
BinaryClassificationMetrics metrics =
  new BinaryClassificationMetrics(scoreAndLabels.rdd());
double auROC = metrics.areaUnderROC();

System.out.println("Loss of each step in training process");
for (double l : loss) {
  System.out.println(l);
}
System.out.println("Area under ROC = " + auROC);
完全なサンプルコードは、Sparkリポジトリの「examples/src/main/java/org/apache/spark/examples/mllib/JavaLBFGSExample.java」にあります。

開発者向けノート

ヘッセ行列は過去の勾配評価から近似的に構築されるため、最適化プロセス中に目的関数を変更することはできません。その結果、Stochastic L-BFGSはミニバッチを単純に使用するだけでは機能しないため、より理解が深まるまで提供されません。

Updater は、実際には勾配降下ステップを計算するために設計されたクラスです。しかし、L-BFGSの正則化項の目的関数の勾配と損失を取得することは、適応ステップサイズなどの勾配降下法専用のロジックを無視することで可能です。これは後で正則化項にリファクタリングしてupdaterを置き換え、正則化とステップ更新のロジックを分離する予定です。