최적화 - RDD 기반 API

최적화 - RDD 기반 API (Optimization – RDD-based API)

머신러닝 모델을 학습한다는 것은 곧 어떤 목적 함수(objective function)를 최소화하는 최적화 문제를 푸는 일이에요. MLlib이 내부적으로 사용하는 대표적인 최적화 방법인 경사 하강법(Gradient Descent), 확률적 경사 하강법(SGD), 그리고 준-뉴턴 방법의 일종인 L-BFGS를 수식과 함께 차근차근 설명해 드릴게요.

출처: 문서

본문

목차 (Table of Contents)

\[ \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}} \]

수학적 설명 (Mathematical description)

경사 하강법 (Gradient descent)

$\min_{\wv \in\R^d} \; f(\wv)$ 형태의 최적화 문제를 푸는 가장 간단한 방법은 경사 하강법이에요. 이러한 일차(first-order) 최적화 방법(경사 하강법과 그 확률적 변형 포함)은 대규모 및 분산 계산에 잘 맞습니다.

경사 하강법은 함수의 현재 지점(즉, 현재 파라미터 값)에서의 도함수(그것을 경사(gradient)라고 해요)의 음수인, 가장 가파른 하강 방향으로 반복적으로 한 걸음씩 이동해 함수의 지역 최솟값을 찾는 것을 목표로 합니다. 목적 함수 $f$가 모든 인자에서 미분 가능하지 않지만 여전히 볼록(convex)하다면, *부분 경사(sub-gradient)*가 경사의 자연스러운 일반화이며 단계 방향의 역할을 맡아요. 어쨌든 $f$의 경사나 부분 경사를 계산하는 것은 비용이 큽니다. 모든 손실 항의 기여를 계산하려면 전체 데이터셋을 한 바퀴 완전히 통과해야 하기 때문이에요.

확률적 경사 하강법 (Stochastic gradient descent, 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}

이것은 특히 자연스러운데, 손실이 각 데이터 포인트에서 나오는 개별 손실들의 평균으로 쓰여지기 때문이에요.

확률적 부분 경사(stochastic subgradient)는 기대값으로 볼 때 원래 목적 함수의 실제 부분 경사를 얻는 벡터의 무작위 선택이에요. 데이토 포인트 하나 $i\in[1..n]$를 균일하게 무작위로 고르면, $\wv$에 대해 $\eqref{eq:regPrimal}$의 확률적 부분 경사를 다음과 같이 얻습니다:

\[ 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$는 정규화기(regularizer) $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}

단계 크기 (Step-size). 파라미터 $\gamma$가 단계 크기인데, 기본 구현에서는 반복 카운터의 제곱근에 따라 감소하도록 고릅니다. 즉 $t$번째 반복에서 $\gamma := \frac{s}{\sqrt{t}}$이며, 입력 파라미터 $s=$ stepSize를 사용해요. 참고로 SGD 방법에 최적의 단계 크기를 고르는 일은 실제로 종종 까다롭고 활발히 연구되는 주제입니다.

경사 (Gradients). spark.mllib에 구현된 머신러닝 방법들의 (부분) 경사 표는 분류 및 회귀 섹션에서 확인할 수 있어요.

근접 업데이트 (Proximal Updates). 단계 방향에서 정규화기의 부분 경사 $R'(\wv)$를 그냥 쓰는 대신, 어떤 경우에는 근접 연산자(proximal operator)를 사용해 개선된 업데이트를 얻을 수 있어요. L1 정규화기의 경우 근접 연산자는 L1Updater에 구현된 소프트 임계값(soft thresholding)으로 주어집니다.

분산 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 (Limited-memory BFGS, L-BFGS)

L-BFGS는 $\min_{\wv \in\R^d} \; f(\wv)$ 형태의 최적화 문제를 풀기 위한 준-뉴턴(quasi-Newton) 방법 계열의 최적화 알고리즘이에요. L-BFGS는 목적 함수의 2차 편도함수를 평가해 Hessian 행렬을 구성하지 않고 목적 함수를 지역적으로 2차 함수로 근사합니다. Hessian 행렬은 이전 경사 평가들로 근사되므로, Newton 방법에서 Hessian 행렬을 명시적으로 계산할 때의 수직 확장성(훈련 피처 수) 문제가 없어요. 그 결과 L-BFGS는 다른 일차 최적화보다 종종 더 빠른 수렴을 달성합니다.

최적화 방법 선택하기

선형 방법은 내부적으로 최적화를 사용하며, spark.mllib의 일부 선형 방법들은 SGD와 L-BFGS 둘 다 지원해요. 서로 다른 최적화 방법은 목적 함수의 특성에 따라 서로 다른 수렴 보장을 가질 수 있으며, 그 문헌을 여기서 모두 다룰 수는 없습니다. 일반적으로 L-BFGS를 사용할 수 있으면 SGD보다 L-BFGS를 권장하는데, L-BFGS가 (더 적은 반복으로) 더 빨리 수렴하는 경향이 있기 때문이에요.

MLlib에서의 구현 (Implementation in MLlib)

경사 하강법과 확률적 경사 하강법

확률적 부분 경사 하강법(SGD)을 포함한 경사 하강법 방법은 MLlib의 저수준 원시 자료로 포함되어 있고, 그 위에 다양한 ML 알고리즘이 개발됩니다. 예를 들어 선형 방법 섹션을 참고하세요.

SGD 클래스 GradientDescent는 다음 파라미터를 설정해요:

  • Gradient는 현재 파라미터 값에서 최적화되는 함수의, 단일 훈련 예제에 대한 확률적 경사를 계산하는 클래스예요. MLlib은 hinge, logistic, least-squares 같은 흔한 손실 함수에 대한 경사 클래스를 포함합니다. 경사 클래스는 입력으로 훈련 예제, 그 레이블, 현재 파라미터 값을 받아요.
  • Updater는 주어진 손실 부분의 경사에 대해 실제 경사 하강 단계(즉 각 반복에서 가중치 업데이트)를 수행하는 클래스예요. 업데이터는 정규화 부분에서의 업데이트도 수행합니다. MLlib은 정규화가 없는 경우와 L1, L2 정규화기에 대한 업데이터를 포함해요.
  • stepSize는 경사 하강의 초기 단계 크기를 나타내는 스칼라 값입니다. MLlib의 모든 업데이터는 t번째 단계에서 stepSize $/ \sqrt{t}$`와 같은 단계 크기를 사용해요.
  • numIterations는 실행할 반복 횟수입니다.
  • regParam은 L1 또는 L2 정규화를 사용할 때의 정규화 파라미터예요.
  • miniBatchFraction은 경사 방향을 계산하기 위해 각 반복에서 샘플링되는 전체 데이터의 비율입니다.
    • 샘플링은 여전히 전체 RDD를 한 번 통과해야 하므로, miniBatchFraction을 줄여도 최적화가 많이 빨라지지 않을 수 있어요. 경사 계산에 선택된 샘플만 사용하므로, 경사가 계산 비용이 클 때 사용자가 가장 큰 속도 향상을 보게 됩니다.

L-BFGS

L-BFGS는 현재 MLlib의 저수준 최적화 원시 자료일 뿐이에요. Linear Regression이나 Logistic Regression 같은 다양한 ML 알고리즘에서 L-BFGS를 사용하려면 LogisticRegressionWithSGD 같은 훈련 API 대신 목적 함수의 경사와 업데이터를 직접 옵티마이저에 전달해야 해요. 아래 예제를 참고하세요. 이는 다음 릴리스에서 개선될 예정입니다.

L1Updater를 사용한 L1 정규화는 동작하지 않아요. L1Updater의 소프트 임계값 로직이 경사 하강법용으로 설계되었기 때문입니다. 개발자 노트를 참고하세요.

L-BFGS 방법 LBFGS.runLBFGS은 다음 파라미터를 가져요:

  • Gradient는 현재 파라미터 값에서 최적화되는 목적 함수의, 단일 훈련 예제에 대한 경사를 계산하는 클래스예요. MLlib은 hinge, logistic, least-squares 같은 흔한 손실 함수에 대한 경사 클래스를 포함합니다. 경사 클래스는 입력으로 훈련 예제, 그 레이블, 현재 파라미터 값을 받아요.
  • Updater는 L-BFGS를 위해 정규화 부분의 목적 함수 경사와 손실을 계산하는 클래스예요. MLlib은 정규화가 없는 경우와 L2 정규화기에 대한 업데이터를 포함해요.
  • numCorrections는 L-BFGS 업데이트에서 사용되는 보정(correction)의 수입니다. 10이 권장됩니다.
  • maxNumIterations는 L-BFGS가 실행될 수 있는 최대 반복 횟수예요.
  • regParam은 정규화를 사용할 때의 정규화 파라미터입니다.
  • convergenceTol은 L-BFGS가 수렴한 것으로 간주될 때 여전히 허용되는 상대 변화량을 제어해요. 이 값은 음수가 아니어야 합니다. 낮은 값일수록 덜 관대해서 일반적으로 더 많은 반복이 실행됩니다. 이 값은 Breeze LBFGS 안에서 평균 개선량과 경사의 노름 둘 다를 살펴봐요.

return은 두 요소를 가진 튜플이에요. 첫 번째 요소는 모든 피처의 가중치를 담은 컬럼 행렬이고, 두 번째 요소는 각 반복마다 계산된 손실을 담은 배열입니다.

다음은 L-BFGS 옵티마이저를 사용해 L2 정규화로 이진 로지스틱 회귀를 훈련하는 예제예요.

Scala:

API에 대한 자세한 내용은 [`LBFGS` Scala 문서](api/scala/org/apache/spark/mllib/optimization/LBFGS.html)와 [`SquaredL2Updater` Scala 문서](api/scala/org/apache/spark/mllib/optimization/SquaredL2Updater.html)를 참고하세요.

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"에서 확인할 수 있어요.

Java:

API에 대한 자세한 내용은 [`LBFGS` Java 문서](api/java/org/apache/spark/mllib/optimization/LBFGS.html)와 [`SquaredL2Updater` Java 문서](api/java/org/apache/spark/mllib/optimization/SquaredL2Updater.html)를 참고하세요.

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"에서 확인할 수 있어요.

개발자 노트 (Developer's notes)

Hessian이 이전 경사 평가들에서 근사적으로 구성되므로, 최적화 과정 중에 목적 함수를 바꿀 수 없습니다. 그 결과 Stochastic L-BFGS는 miniBatch를 그냥 사용해서는 원시적으로 동작하지 않아요. 따라서 더 잘 이해하기 전까지는 이를 제공하지 않습니다.

Updater는 원래 경사 하강을 위해 설계된 클래스로, 실제 경사 하강 단계를 계산해요. 하지만 적응 단계 크기 같은 경사 하강 전용 로직 부분을 무시함으로써 L-BFGS를 위한 정규화의 목적 함수 경사와 손실을 취할 수 있습니다. 나중에 이를 regularizer로 리팩터링해 updater를 대체하고 정규화와 단계 업데이트 로직을 분리할 예정이에요.

더 알아보기 (Learn more)