次元削減 - RDDベースAPI
次元削減 は、考慮する変数の数を減らすプロセスです。これは、生のノイズの多い特徴量から潜在的な特徴量を抽出したり、構造を維持しながらデータを圧縮するために使用できます。spark.mllib は、RowMatrix クラスでの次元削減をサポートしています。
特異値分解 (SVD)
特異値分解 (SVD) は、行列を3つの行列に因数分解します：$U$、$\Sigma$、および$V$。これらは以下のように表されます。
\[ A = U \Sigma V^T, \]
ここで
- $U$ は、列が左特異ベクトルと呼ばれる直交行列です。
- $\Sigma$ は、対角線が降順の非負の対角線を持つ対角行列であり、その対角線は特異値と呼ばれます。
- $V$ は、列が右特異ベクトルと呼ばれる直交行列です。
大規模な行列の場合、通常は完全な分解ではなく、上位の特異値とその関連する特異ベクトルのみが必要になります。これにより、ストレージの節約、ノイズ除去、および行列の低ランク構造の復元が可能になります。
上位$k$個の特異値を保持する場合、結果として得られる低ランク行列の次元は次のようになります。
$U$:$m \times k$,$\Sigma$:$k \times k$,$V$:$n \times k$.
パフォーマンス
ここで、$n$ は $m$ より小さいと仮定します。特異値と右特異ベクトルは、グラム行列 $A^T A$ の固有値と固有ベクトルから導出されます。左特異ベクトル $U$ を格納する行列は、ユーザーが computeU パラメータを介して要求した場合、行列乗算 $U = A (V S^{-1})$ を介して計算されます。使用する実際の方法は、計算コストに基づいて自動的に決定されます。
- $n$ が小さい ($n < 100$) か、$k$ が $n$ に比べて大きい ($k > n / 2$) 場合、まずグラム行列を計算し、次にドライバ上でその上位固有値と固有ベクトルをローカルに計算します。これには、各エグゼキュータとドライバ上で $O(n^2)$ のストレージ、およびドライバ上で $O(n^2 k)$ の時間がかかります。
- それ以外の場合は、$(A^T A) v$ を分散的に計算し、ARPACK に送信して、ドライバノードで $(A^T A)$ の上位固有値と固有ベクトルを計算します。これには、$O(k)$ 回のパス、各エグゼキュータ上で $O(n)$ のストレージ、およびドライバ上で $O(n k)$ のストレージが必要です。
SVDの例
spark.mllib は、RowMatrix クラスで提供される行指向行列へのSVD機能を提供します。
APIの詳細については、SingularValueDecomposition Pythonドキュメント を参照してください。
from pyspark.mllib.linalg import Vectors
from pyspark.mllib.linalg.distributed import RowMatrix
rows = sc.parallelize([
Vectors.sparse(5, {1: 1.0, 3: 7.0}),
Vectors.dense(2.0, 0.0, 3.0, 4.0, 5.0),
Vectors.dense(4.0, 0.0, 0.0, 6.0, 7.0)
])
mat = RowMatrix(rows)
# Compute the top 5 singular values and corresponding singular vectors.
svd = mat.computeSVD(5, computeU=True)
U = svd.U # The U factor is a RowMatrix.
s = svd.s # The singular values are stored in a local dense vector.
V = svd.V # The V factor is a local dense matrix.完全なサンプルコードは、Sparkリポジトリの "examples/src/main/python/mllib/svd_example.py" にあります。
U が IndexedRowMatrix として定義されている場合、同じコードが IndexedRowMatrix に適用されます。
APIの詳細については、SingularValueDecomposition Scalaドキュメント を参照してください。
import org.apache.spark.mllib.linalg.Matrix
import org.apache.spark.mllib.linalg.SingularValueDecomposition
import org.apache.spark.mllib.linalg.Vector
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.linalg.distributed.RowMatrix
val data = Array(
Vectors.sparse(5, Seq((1, 1.0), (3, 7.0))),
Vectors.dense(2.0, 0.0, 3.0, 4.0, 5.0),
Vectors.dense(4.0, 0.0, 0.0, 6.0, 7.0))
val rows = sc.parallelize(data)
val mat: RowMatrix = new RowMatrix(rows)
// Compute the top 5 singular values and corresponding singular vectors.
val svd: SingularValueDecomposition[RowMatrix, Matrix] = mat.computeSVD(5, computeU = true)
val U: RowMatrix = svd.U // The U factor is a RowMatrix.
val s: Vector = svd.s // The singular values are stored in a local dense vector.
val V: Matrix = svd.V // The V factor is a local dense matrix.完全なサンプルコードは、Sparkリポジトリの "examples/src/main/scala/org/apache/spark/examples/mllib/SVDExample.scala" にあります。
U が IndexedRowMatrix として定義されている場合、同じコードが IndexedRowMatrix に適用されます。
APIの詳細については、SingularValueDecomposition Javaドキュメント を参照してください。
import java.util.Arrays;
import java.util.List;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.mllib.linalg.Matrix;
import org.apache.spark.mllib.linalg.SingularValueDecomposition;
import org.apache.spark.mllib.linalg.Vector;
import org.apache.spark.mllib.linalg.Vectors;
import org.apache.spark.mllib.linalg.distributed.RowMatrix;
List<Vector> data = Arrays.asList(
Vectors.sparse(5, new int[] {1, 3}, new double[] {1.0, 7.0}),
Vectors.dense(2.0, 0.0, 3.0, 4.0, 5.0),
Vectors.dense(4.0, 0.0, 0.0, 6.0, 7.0)
);
JavaRDD<Vector> rows = jsc.parallelize(data);
// Create a RowMatrix from JavaRDD<Vector>.
RowMatrix mat = new RowMatrix(rows.rdd());
// Compute the top 5 singular values and corresponding singular vectors.
SingularValueDecomposition<RowMatrix, Matrix> svd = mat.computeSVD(5, true, 1.0E-9d);
RowMatrix U = svd.U(); // The U factor is a RowMatrix.
Vector s = svd.s(); // The singular values are stored in a local dense vector.
Matrix V = svd.V(); // The V factor is a local dense matrix.完全なサンプルコードは、Sparkリポジトリの "examples/src/main/java/org/apache/spark/examples/mllib/JavaSVDExample.java" にあります。
U が IndexedRowMatrix として定義されている場合、同じコードが IndexedRowMatrix に適用されます。
主成分分析 (PCA)
主成分分析 (PCA) は、最初の座標が可能な限り最大の分散を持ち、その後の各座標が順番に可能な限り最大の分散を持つような回転を見つける統計的手法です。回転行列の列は主成分と呼ばれます。PCAは次元削減で広く使用されています。
spark.mllib は、行指向形式で格納された細長い行列と任意のベクトルに対するPCAをサポートしています。
次のコードは、RowMatrix で主成分を計算し、それらを使用してベクトルを低次元空間に投影する方法を示しています。
APIの詳細については、RowMatrix Pythonドキュメント を参照してください。
from pyspark.mllib.linalg import Vectors
from pyspark.mllib.linalg.distributed import RowMatrix
rows = sc.parallelize([
Vectors.sparse(5, {1: 1.0, 3: 7.0}),
Vectors.dense(2.0, 0.0, 3.0, 4.0, 5.0),
Vectors.dense(4.0, 0.0, 0.0, 6.0, 7.0)
])
mat = RowMatrix(rows)
# Compute the top 4 principal components.
# Principal components are stored in a local dense matrix.
pc = mat.computePrincipalComponents(4)
# Project the rows to the linear space spanned by the top 4 principal components.
projected = mat.multiply(pc)完全なサンプルコードは、Sparkリポジトリの "examples/src/main/python/mllib/pca_rowmatrix_example.py" にあります。
次のコードは、RowMatrix で主成分を計算し、それらを使用してベクトルを低次元空間に投影する方法を示しています。
APIの詳細については、RowMatrix Scalaドキュメント を参照してください。
import org.apache.spark.mllib.linalg.Matrix
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.linalg.distributed.RowMatrix
val data = Array(
Vectors.sparse(5, Seq((1, 1.0), (3, 7.0))),
Vectors.dense(2.0, 0.0, 3.0, 4.0, 5.0),
Vectors.dense(4.0, 0.0, 0.0, 6.0, 7.0))
val rows = sc.parallelize(data)
val mat: RowMatrix = new RowMatrix(rows)
// Compute the top 4 principal components.
// Principal components are stored in a local dense matrix.
val pc: Matrix = mat.computePrincipalComponents(4)
// Project the rows to the linear space spanned by the top 4 principal components.
val projected: RowMatrix = mat.multiply(pc)完全なサンプルコードは、Sparkリポジトリの "examples/src/main/scala/org/apache/spark/examples/mllib/PCAOnRowMatrixExample.scala" にあります。
次のコードは、ソースベクトルで主成分を計算し、関連するラベルを維持しながら、それらを使用してベクトルを低次元空間に投影する方法を示しています。
APIの詳細については、PCA Scalaドキュメント を参照してください。
import org.apache.spark.mllib.feature.PCA
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.rdd.RDD
val data: RDD[LabeledPoint] = sc.parallelize(Seq(
new LabeledPoint(0, Vectors.dense(1, 0, 0, 0, 1)),
new LabeledPoint(1, Vectors.dense(1, 1, 0, 1, 0)),
new LabeledPoint(1, Vectors.dense(1, 1, 0, 0, 0)),
new LabeledPoint(0, Vectors.dense(1, 0, 0, 0, 0)),
new LabeledPoint(1, Vectors.dense(1, 1, 0, 0, 0))))
// Compute the top 5 principal components.
val pca = new PCA(5).fit(data.map(_.features))
// Project vectors to the linear space spanned by the top 5 principal
// components, keeping the label
val projected = data.map(p => p.copy(features = pca.transform(p.features)))完全なサンプルコードは、Sparkリポジトリの "examples/src/main/scala/org/apache/spark/examples/mllib/PCAOnSourceVectorExample.scala" にあります。
次のコードは、RowMatrix で主成分を計算し、それらを使用してベクトルを低次元空間に投影する方法を示しています。
APIの詳細については、RowMatrix Javaドキュメント を参照してください。
import java.util.Arrays;
import java.util.List;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.mllib.linalg.Matrix;
import org.apache.spark.mllib.linalg.Vector;
import org.apache.spark.mllib.linalg.Vectors;
import org.apache.spark.mllib.linalg.distributed.RowMatrix;
List<Vector> data = Arrays.asList(
Vectors.sparse(5, new int[] {1, 3}, new double[] {1.0, 7.0}),
Vectors.dense(2.0, 0.0, 3.0, 4.0, 5.0),
Vectors.dense(4.0, 0.0, 0.0, 6.0, 7.0)
);
JavaRDD<Vector> rows = jsc.parallelize(data);
// Create a RowMatrix from JavaRDD<Vector>.
RowMatrix mat = new RowMatrix(rows.rdd());
// Compute the top 4 principal components.
// Principal components are stored in a local dense matrix.
Matrix pc = mat.computePrincipalComponents(4);
// Project the rows to the linear space spanned by the top 4 principal components.
RowMatrix projected = mat.multiply(pc);完全なサンプルコードは、Sparkリポジトリの "examples/src/main/java/org/apache/spark/examples/mllib/JavaPCAExample.java" にあります。