データ型 - RDDベースのAPI

ローカルベクトル
ラベル付きポイント
ローカル行列
分散行列

MLlibは、単一のマシンに格納されたローカルベクトルと行列、および1つ以上のRDDによってバックアップされた分散行列をサポートしています。ローカルベクトルとローカル行列は、パブリックインターフェースとして機能する単純なデータモデルです。基盤となる線形代数演算は、Breezeによって提供されます。教師あり学習で使用されるトレーニング例は、MLlibでは「ラベル付きポイント」と呼ばれます。

ローカルベクトル

ローカルベクトルは、整数型の0ベースのインデックスと倍精度型の値を持ち、単一のマシンに格納されます。MLlibは、密と疎の2種類のローカルベクトルをサポートしています。密ベクトルは、そのエントリ値を表すdouble配列によってバックアップされ、疎ベクトルは、インデックスと値の2つの並列配列によってバックアップされます。たとえば、ベクトル(1.0, 0.0, 3.0)は、密形式では[1.0, 0.0, 3.0]、疎形式では(3, [0, 2], [1.0, 3.0])として表すことができます。ここで、3はベクトルのサイズです。

MLlibは、次の型を密ベクトルとして認識します。

NumPyのarray
Pythonのリスト（例：[1, 2, 3]）

また、次の型を疎ベクトルとして認識します。

MLlibのSparseVector。
SciPyのcsc_matrix（単一の列を持つ）

効率のためにはリストよりもNumPy配列を使用し、疎ベクトルを作成するにはVectorsに実装されているファクトリメソッドを使用することをお勧めします。

APIの詳細については、Vectors Pythonドキュメントを参照してください。

import numpy as np
import scipy.sparse as sps
from pyspark.mllib.linalg import Vectors

# Use a NumPy array as a dense vector.
dv1 = np.array([1.0, 0.0, 3.0])
# Use a Python list as a dense vector.
dv2 = [1.0, 0.0, 3.0]
# Create a SparseVector.
sv1 = Vectors.sparse(3, [0, 2], [1.0, 3.0])
# Use a single-column SciPy csc_matrix as a sparse vector.
sv2 = sps.csc_matrix((np.array([1.0, 3.0]), np.array([0, 2]), np.array([0, 2])), shape=(3, 1))

ローカルベクトルのベースクラスはVectorであり、DenseVectorとSparseVectorの2つの実装を提供しています。ローカルベクトルを作成するには、Vectorsに実装されているファクトリメソッドを使用することをお勧めします。

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

import org.apache.spark.mllib.linalg.{Vector, Vectors}

// Create a dense vector (1.0, 0.0, 3.0).
val dv: Vector = Vectors.dense(1.0, 0.0, 3.0)
// Create a sparse vector (1.0, 0.0, 3.0) by specifying its indices and values corresponding to nonzero entries.
val sv1: Vector = Vectors.sparse(3, Array(0, 2), Array(1.0, 3.0))
// Create a sparse vector (1.0, 0.0, 3.0) by specifying its nonzero entries.
val sv2: Vector = Vectors.sparse(3, Seq((0, 1.0), (2, 3.0)))

注： Scalaはデフォルトでscala.collection.immutable.Vectorをインポートするため、MLlibのVectorを使用するには、org.apache.spark.mllib.linalg.Vectorを明示的にインポートする必要があります。

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

import org.apache.spark.mllib.linalg.Vector;
import org.apache.spark.mllib.linalg.Vectors;

// Create a dense vector (1.0, 0.0, 3.0).
Vector dv = Vectors.dense(1.0, 0.0, 3.0);
// Create a sparse vector (1.0, 0.0, 3.0) by specifying its indices and values corresponding to nonzero entries.
Vector sv = Vectors.sparse(3, new int[] {0, 2}, new double[] {1.0, 3.0});

ラベル付きポイント

ラベル付きポイントは、密または疎のローカルベクトルであり、ラベル/応答に関連付けられています。MLlibでは、ラベル付きポイントは教師あり学習アルゴリズムで使用されます。ラベルを格納するためにdoubleを使用しているため、回帰と分類の両方でラベル付きポイントを使用できます。二項分類の場合、ラベルは0（負）または1（正）である必要があります。多クラス分類の場合、ラベルはゼロから始まるクラスインデックスである必要があります：0、1、2、...。

ラベル付きポイントは、LabeledPointで表されます。

APIの詳細については、LabeledPoint Pythonドキュメントを参照してください。

from pyspark.mllib.linalg import SparseVector
from pyspark.mllib.regression import LabeledPoint

# Create a labeled point with a positive label and a dense feature vector.
pos = LabeledPoint(1.0, [1.0, 0.0, 3.0])

# Create a labeled point with a negative label and a sparse feature vector.
neg = LabeledPoint(0.0, SparseVector(3, [0, 2], [1.0, 3.0]))

ラベル付きポイントは、ケースクラスLabeledPointで表されます。

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

import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.regression.LabeledPoint

// Create a labeled point with a positive label and a dense feature vector.
val pos = LabeledPoint(1.0, Vectors.dense(1.0, 0.0, 3.0))

// Create a labeled point with a negative label and a sparse feature vector.
val neg = LabeledPoint(0.0, Vectors.sparse(3, Array(0, 2), Array(1.0, 3.0)))

ラベル付きポイントは、LabeledPointで表されます。

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

import org.apache.spark.mllib.linalg.Vectors;
import org.apache.spark.mllib.regression.LabeledPoint;

// Create a labeled point with a positive label and a dense feature vector.
LabeledPoint pos = new LabeledPoint(1.0, Vectors.dense(1.0, 0.0, 3.0));

// Create a labeled point with a negative label and a sparse feature vector.
LabeledPoint neg = new LabeledPoint(0.0, Vectors.sparse(3, new int[] {0, 2}, new double[] {1.0, 3.0}));

スパースデータ

実際には、疎なトレーニングデータを使用することが非常に一般的です。MLlibは、LIBSVM形式で保存されたトレーニング例の読み込みをサポートしています。これは、LIBSVMおよびLIBLINEARで使用されるデフォルト形式です。これはテキスト形式であり、各行は次の形式を使用してラベル付きの疎な特徴ベクトルを表します。

label index1:value1 index2:value2 ...

ここで、インデックスは1ベースで昇順です。ロード後、特徴インデックスは0ベースに変換されます。

MLUtils.loadLibSVMFileは、LIBSVM形式で保存されたトレーニング例を読み込みます。

APIの詳細については、MLUtils Pythonドキュメントを参照してください。

from pyspark.mllib.util import MLUtils

examples = MLUtils.loadLibSVMFile(sc, "data/mllib/sample_libsvm_data.txt")

MLUtils.loadLibSVMFileは、LIBSVM形式で保存されたトレーニング例を読み込みます。

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

import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.mllib.util.MLUtils
import org.apache.spark.rdd.RDD

val examples: RDD[LabeledPoint] = MLUtils.loadLibSVMFile(sc, "data/mllib/sample_libsvm_data.txt")

MLUtils.loadLibSVMFileは、LIBSVM形式で保存されたトレーニング例を読み込みます。

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

import org.apache.spark.mllib.regression.LabeledPoint;
import org.apache.spark.mllib.util.MLUtils;
import org.apache.spark.api.java.JavaRDD;

JavaRDD<LabeledPoint> examples = 
  MLUtils.loadLibSVMFile(jsc.sc(), "data/mllib/sample_libsvm_data.txt").toJavaRDD();

ローカル行列

ローカル行列は、整数型の行と列のインデックスと倍精度型の値を持ち、単一のマシンに格納されます。MLlibは、エントリ値が列優先順で単一のdouble配列に格納される密行列と、非ゼロのエントリ値が列優先順でCompressed Sparse Column（CSC）形式で格納される疎行列をサポートしています。たとえば、次の密行列 $\begin{pmatrix} 1.0 & 2.0 \\ 3.0 & 4.0 \\ 5.0 & 6.0 \end{pmatrix}$ は、行列サイズ(3, 2)とともに1次元配列[1.0, 3.0, 5.0, 2.0, 4.0, 6.0]に格納されます。

ローカル行列のベースクラスはMatrixであり、DenseMatrixとSparseMatrixの2つの実装を提供しています。ローカル行列を作成するには、Matricesに実装されているファクトリメソッドを使用することをお勧めします。MLlibのローカル行列は、列優先順で格納されることに注意してください。

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

from pyspark.mllib.linalg import Matrix, Matrices

# Create a dense matrix ((1.0, 2.0), (3.0, 4.0), (5.0, 6.0))
dm2 = Matrices.dense(3, 2, [1, 3, 5, 2, 4, 6])

# Create a sparse matrix ((9.0, 0.0), (0.0, 8.0), (0.0, 6.0))
sm = Matrices.sparse(3, 2, [0, 1, 3], [0, 2, 1], [9, 6, 8])

ローカル行列の基本クラスはMatrixであり、2つの実装を提供しています。DenseMatrixとSparseMatrixです。ローカル行列を作成するには、Matricesに実装されているファクトリメソッドを使用することを推奨します。MLlibのローカル行列は、列優先の順序で格納されることに注意してください。

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

import org.apache.spark.mllib.linalg.{Matrix, Matrices}

// Create a dense matrix ((1.0, 2.0), (3.0, 4.0), (5.0, 6.0))
val dm: Matrix = Matrices.dense(3, 2, Array(1.0, 3.0, 5.0, 2.0, 4.0, 6.0))

// Create a sparse matrix ((9.0, 0.0), (0.0, 8.0), (0.0, 6.0))
val sm: Matrix = Matrices.sparse(3, 2, Array(0, 1, 3), Array(0, 2, 1), Array(9, 6, 8))

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

import org.apache.spark.mllib.linalg.Matrix;
import org.apache.spark.mllib.linalg.Matrices;

// Create a dense matrix ((1.0, 2.0), (3.0, 4.0), (5.0, 6.0))
Matrix dm = Matrices.dense(3, 2, new double[] {1.0, 3.0, 5.0, 2.0, 4.0, 6.0});

// Create a sparse matrix ((9.0, 0.0), (0.0, 8.0), (0.0, 6.0))
Matrix sm = Matrices.sparse(3, 2, new int[] {0, 1, 3}, new int[] {0, 2, 1}, new double[] {9, 6, 8});

分散行列

分散行列は、long型の行インデックスと列インデックス、およびdouble型の値を持ち、1つ以上のRDDに分散して格納されます。大規模な分散行列を格納するには、適切な形式を選択することが非常に重要です。分散行列を別の形式に変換するには、非常にコストのかかるグローバルシャッフルが必要になる場合があります。これまでに4種類の分散行列が実装されています。

基本型はRowMatrixと呼ばれます。RowMatrixは、意味のある行インデックスを持たない、行指向の分散行列です。例えば、特徴ベクトルのコレクションです。各行はローカルベクトルであり、行のRDDによってバックアップされます。RowMatrixの場合、列数はそれほど大きくないと想定しているため、単一のローカルベクトルをドライバーに合理的に通信でき、単一のノードを使用して格納/操作することもできます。IndexedRowMatrixはRowMatrixに似ていますが、行インデックスがあり、行の識別や結合の実行に使用できます。CoordinateMatrixは、座標リスト（COO）形式で格納された分散行列であり、エントリのRDDによってバックアップされます。BlockMatrixは、(Int, Int, Matrix)のタプルであるMatrixBlockのRDDによってバックアップされる分散行列です。

注

分散行列の基になるRDDは、行列のサイズをキャッシュするため、決定的である必要があります。一般に、非決定的なRDDを使用すると、エラーが発生する可能性があります。

RowMatrix

RowMatrixは、意味のある行インデックスを持たない、行指向の分散行列であり、各行がローカルベクトルである行のRDDによってバックアップされます。各行はローカルベクトルで表されるため、列数は整数範囲によって制限されますが、実際にははるかに小さいはずです。

RowMatrixは、ベクトルのRDDから作成できます。

APIの詳細については、RowMatrix Pythonドキュメントを参照してください。

from pyspark.mllib.linalg.distributed import RowMatrix

# Create an RDD of vectors.
rows = sc.parallelize([[1, 2, 3], [4, 5, 6], [7, 8, 9], [10, 11, 12]])

# Create a RowMatrix from an RDD of vectors.
mat = RowMatrix(rows)

# Get its size.
m = mat.numRows()  # 4
n = mat.numCols()  # 3

# Get the rows as an RDD of vectors again.
rowsRDD = mat.rows

RowMatrixは、RDD[Vector]インスタンスから作成できます。その後、列の要約統計量と分解を計算できます。QR分解は、Qが直交行列、Rが上三角行列である形式A = QRです。特異値分解（SVD）と主成分分析（PCA）については、次元削減を参照してください。

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

import org.apache.spark.mllib.linalg.Vector
import org.apache.spark.mllib.linalg.distributed.RowMatrix

val rows: RDD[Vector] = ... // an RDD of local vectors
// Create a RowMatrix from an RDD[Vector].
val mat: RowMatrix = new RowMatrix(rows)

// Get its size.
val m = mat.numRows()
val n = mat.numCols()

// QR decomposition 
val qrResult = mat.tallSkinnyQR(true)

RowMatrixは、JavaRDD<Vector>インスタンスから作成できます。その後、列の要約統計量を計算できます。

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

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.mllib.linalg.Vector;
import org.apache.spark.mllib.linalg.distributed.RowMatrix;

JavaRDD<Vector> rows = ... // a JavaRDD of local vectors
// Create a RowMatrix from a JavaRDD<Vector>.
RowMatrix mat = new RowMatrix(rows.rdd());

// Get its size.
long m = mat.numRows();
long n = mat.numCols();

// QR decomposition 
QRDecomposition<RowMatrix, Matrix> result = mat.tallSkinnyQR(true);

IndexedRowMatrix

IndexedRowMatrixはRowMatrixに似ていますが、意味のある行インデックスがあります。インデックス付き行のRDDによってバックアップされるため、各行はインデックス（long型）とローカルベクトルで表されます。

IndexedRowMatrixは、IndexedRowのRDDから作成できます。ここで、IndexedRowは、(long, vector)のラッパーです。IndexedRowMatrixは、行インデックスを削除することでRowMatrixに変換できます。

APIの詳細については、IndexedRowMatrix Pythonドキュメントを参照してください。

from pyspark.mllib.linalg.distributed import IndexedRow, IndexedRowMatrix

# Create an RDD of indexed rows.
#   - This can be done explicitly with the IndexedRow class:
indexedRows = sc.parallelize([IndexedRow(0, [1, 2, 3]),
                              IndexedRow(1, [4, 5, 6]),
                              IndexedRow(2, [7, 8, 9]),
                              IndexedRow(3, [10, 11, 12])])
#   - or by using (long, vector) tuples:
indexedRows = sc.parallelize([(0, [1, 2, 3]), (1, [4, 5, 6]),
                              (2, [7, 8, 9]), (3, [10, 11, 12])])

# Create an IndexedRowMatrix from an RDD of IndexedRows.
mat = IndexedRowMatrix(indexedRows)

# Get its size.
m = mat.numRows()  # 4
n = mat.numCols()  # 3

# Get the rows as an RDD of IndexedRows.
rowsRDD = mat.rows

# Convert to a RowMatrix by dropping the row indices.
rowMat = mat.toRowMatrix()

IndexedRowMatrixは、RDD[IndexedRow]インスタンスから作成できます。ここで、IndexedRowは(Long, Vector)のラッパーです。IndexedRowMatrixは、行インデックスを削除することでRowMatrixに変換できます。

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

import org.apache.spark.mllib.linalg.distributed.{IndexedRow, IndexedRowMatrix, RowMatrix}

val rows: RDD[IndexedRow] = ... // an RDD of indexed rows
// Create an IndexedRowMatrix from an RDD[IndexedRow].
val mat: IndexedRowMatrix = new IndexedRowMatrix(rows)

// Get its size.
val m = mat.numRows()
val n = mat.numCols()

// Drop its row indices.
val rowMat: RowMatrix = mat.toRowMatrix()

IndexedRowMatrixは、JavaRDD<IndexedRow>インスタンスから作成できます。ここで、IndexedRowは、(long, Vector)のラッパーです。IndexedRowMatrixは、行インデックスを削除することでRowMatrixに変換できます。

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

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.mllib.linalg.distributed.IndexedRow;
import org.apache.spark.mllib.linalg.distributed.IndexedRowMatrix;
import org.apache.spark.mllib.linalg.distributed.RowMatrix;

JavaRDD<IndexedRow> rows = ... // a JavaRDD of indexed rows
// Create an IndexedRowMatrix from a JavaRDD<IndexedRow>.
IndexedRowMatrix mat = new IndexedRowMatrix(rows.rdd());

// Get its size.
long m = mat.numRows();
long n = mat.numCols();

// Drop its row indices.
RowMatrix rowMat = mat.toRowMatrix();

CoordinateMatrix

CoordinateMatrixは、エントリのRDDによってバックアップされる分散行列です。各エントリは、(i: Long, j: Long, value: Double)のタプルです。ここで、iは行インデックス、jは列インデックス、valueはエントリの値です。CoordinateMatrixは、行列の両方の次元が非常に大きく、行列が非常に疎である場合にのみ使用する必要があります。

CoordinateMatrixは、MatrixEntryエントリのRDDから作成できます。ここで、MatrixEntryは(long, long, float)のラッパーです。CoordinateMatrixは、toRowMatrixを呼び出すことでRowMatrixに、またはtoIndexedRowMatrixを呼び出すことで疎な行を持つIndexedRowMatrixに変換できます。

APIの詳細については、CoordinateMatrix Pythonドキュメントを参照してください。

from pyspark.mllib.linalg.distributed import CoordinateMatrix, MatrixEntry

# Create an RDD of coordinate entries.
#   - This can be done explicitly with the MatrixEntry class:
entries = sc.parallelize([MatrixEntry(0, 0, 1.2), MatrixEntry(1, 0, 2.1), MatrixEntry(2, 1, 3.7)])
#   - or using (long, long, float) tuples:
entries = sc.parallelize([(0, 0, 1.2), (1, 0, 2.1), (2, 1, 3.7)])

# Create a CoordinateMatrix from an RDD of MatrixEntries.
mat = CoordinateMatrix(entries)

# Get its size.
m = mat.numRows()  # 3
n = mat.numCols()  # 2

# Get the entries as an RDD of MatrixEntries.
entriesRDD = mat.entries

# Convert to a RowMatrix.
rowMat = mat.toRowMatrix()

# Convert to an IndexedRowMatrix.
indexedRowMat = mat.toIndexedRowMatrix()

# Convert to a BlockMatrix.
blockMat = mat.toBlockMatrix()

CoordinateMatrixは、RDD[MatrixEntry]インスタンスから作成できます。ここで、MatrixEntryは(Long, Long, Double)のラッパーです。CoordinateMatrixは、toIndexedRowMatrixを呼び出すことで、疎な行を持つIndexedRowMatrixに変換できます。CoordinateMatrixのその他の計算は、現在サポートされていません。

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

import org.apache.spark.mllib.linalg.distributed.{CoordinateMatrix, MatrixEntry}

val entries: RDD[MatrixEntry] = ... // an RDD of matrix entries
// Create a CoordinateMatrix from an RDD[MatrixEntry].
val mat: CoordinateMatrix = new CoordinateMatrix(entries)

// Get its size.
val m = mat.numRows()
val n = mat.numCols()

// Convert it to an IndexRowMatrix whose rows are sparse vectors.
val indexedRowMatrix = mat.toIndexedRowMatrix()

CoordinateMatrixは、JavaRDD<MatrixEntry>インスタンスから作成できます。ここで、MatrixEntryは(long, long, double)のラッパーです。CoordinateMatrixは、toIndexedRowMatrixを呼び出すことで、疎な行を持つIndexedRowMatrixに変換できます。CoordinateMatrixのその他の計算は、現在サポートされていません。

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

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.mllib.linalg.distributed.CoordinateMatrix;
import org.apache.spark.mllib.linalg.distributed.IndexedRowMatrix;
import org.apache.spark.mllib.linalg.distributed.MatrixEntry;

JavaRDD<MatrixEntry> entries = ... // a JavaRDD of matrix entries
// Create a CoordinateMatrix from a JavaRDD<MatrixEntry>.
CoordinateMatrix mat = new CoordinateMatrix(entries.rdd());

// Get its size.
long m = mat.numRows();
long n = mat.numCols();

// Convert it to an IndexRowMatrix whose rows are sparse vectors.
IndexedRowMatrix indexedRowMatrix = mat.toIndexedRowMatrix();

BlockMatrix

BlockMatrixは、MatrixBlockのRDDによってバックアップされる分散行列です。ここで、MatrixBlockは((Int, Int), Matrix)のタプルであり、(Int, Int)はブロックのインデックス、Matrixは指定されたインデックスのサブ行列であり、サイズはrowsPerBlock x colsPerBlockです。BlockMatrixは、別のBlockMatrixとのaddやmultiplyなどのメソッドをサポートしています。BlockMatrixには、BlockMatrixが正しく設定されているかどうかを確認するために使用できるヘルパー関数validateもあります。

BlockMatrixは、サブ行列ブロックのRDDから作成できます。ここで、サブ行列ブロックは((blockRowIndex, blockColIndex), sub-matrix)タプルです。

APIの詳細については、BlockMatrix Pythonドキュメントを参照してください。

from pyspark.mllib.linalg import Matrices
from pyspark.mllib.linalg.distributed import BlockMatrix

# Create an RDD of sub-matrix blocks.
blocks = sc.parallelize([((0, 0), Matrices.dense(3, 2, [1, 2, 3, 4, 5, 6])),
                         ((1, 0), Matrices.dense(3, 2, [7, 8, 9, 10, 11, 12]))])

# Create a BlockMatrix from an RDD of sub-matrix blocks.
mat = BlockMatrix(blocks, 3, 2)

# Get its size.
m = mat.numRows()  # 6
n = mat.numCols()  # 2

# Get the blocks as an RDD of sub-matrix blocks.
blocksRDD = mat.blocks

# Convert to a LocalMatrix.
localMat = mat.toLocalMatrix()

# Convert to an IndexedRowMatrix.
indexedRowMat = mat.toIndexedRowMatrix()

# Convert to a CoordinateMatrix.
coordinateMat = mat.toCoordinateMatrix()

BlockMatrix は、IndexedRowMatrix または CoordinateMatrix から toBlockMatrix を呼び出すことで最も簡単に作成できます。toBlockMatrix はデフォルトで 1024 x 1024 のサイズのブロックを作成します。ユーザーは、toBlockMatrix(rowsPerBlock, colsPerBlock) を通して値を指定することでブロックサイズを変更できます。

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

import org.apache.spark.mllib.linalg.distributed.{BlockMatrix, CoordinateMatrix, MatrixEntry}

val entries: RDD[MatrixEntry] = ... // an RDD of (i, j, v) matrix entries
// Create a CoordinateMatrix from an RDD[MatrixEntry].
val coordMat: CoordinateMatrix = new CoordinateMatrix(entries)
// Transform the CoordinateMatrix to a BlockMatrix
val matA: BlockMatrix = coordMat.toBlockMatrix().cache()

// Validate whether the BlockMatrix is set up properly. Throws an Exception when it is not valid.
// Nothing happens if it is valid.
matA.validate()

// Calculate A^T A.
val ata = matA.transpose.multiply(matA)

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

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.mllib.linalg.distributed.BlockMatrix;
import org.apache.spark.mllib.linalg.distributed.CoordinateMatrix;
import org.apache.spark.mllib.linalg.distributed.IndexedRowMatrix;

JavaRDD<MatrixEntry> entries = ... // a JavaRDD of (i, j, v) Matrix Entries
// Create a CoordinateMatrix from a JavaRDD<MatrixEntry>.
CoordinateMatrix coordMat = new CoordinateMatrix(entries.rdd());
// Transform the CoordinateMatrix to a BlockMatrix
BlockMatrix matA = coordMat.toBlockMatrix().cache();

// Validate whether the BlockMatrix is set up properly. Throws an Exception when it is not valid.
// Nothing happens if it is valid.
matA.validate();

// Calculate A^T A.
BlockMatrix ata = matA.transpose().multiply(matA);

MLlib: メインガイド

MLlib: RDDベースのAPIガイド

データ型 - RDDベースのAPI

ローカルベクトル

ラベル付きポイント

ローカル行列

分散行列

RowMatrix

IndexedRowMatrix

CoordinateMatrix

BlockMatrix