返回 Spark MLlib入门教程

MLLib提供了一序列基本数据类型以支持底层的机器学习算法。主要的数据内心包括：本地向量、标注点（Labeled Point）、本地矩阵、分布式矩阵等。单机模式存储的本地向量与矩阵，以及基于一个或多个RDD的分布式矩阵。其中本地向量与本地矩阵作为公共接口提供简单数据模型，底层的线性代数操作由Breeze库和jblas库提供。标注点类型用来表示监督学习（Supervised Learning）中的一个训练样本。

在正式学习机器学习算法之前，让我们先了解下这些数据类型的用法。

##一、本地向量（Local Vector）

本地向量存储在单机上，其拥有整型、从0开始的索引值以及浮点型的元素值。MLlib提供了两种类型的本地向量，稠密向量DenseVector和稀疏向量SparseVector。稠密向量使用一个双精度浮点型数组来表示其中每一维元素，而稀疏向量则是基于一个整型索引数组和一个双精度浮点型的值数组。例如，向量(1.0, 0.0, 3.0)的稠密向量表示形式是[1.0,0.0,3.0]，而稀疏向量形式则是(3, [0,2], [1.0, 3.0])，其中，3是向量的长度，[0,2]是向量中非0维度的索引值，表示位置为0、2的两个元素为非零值，而[1.0, 3.0]则是按索引排列的数组元素值。

所有本地向量都以org.apache.spark.mllib.linalg.Vector为基类，DenseVector和SparseVector分别是它的两个实现类，故推荐使用Vectors工具类下定义的工厂方法来创建本地向量，请看如下实例（假设在Spark-shell中运行，下同）：

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

// 创建一个稠密本地向量
scala> val dv: Vector = Vectors.dense(2.0, 0.0, 8.0)
dv: org.apache.spark.mllib.linalg.Vector = [2.0,0.0,8.0]
// 创建一个稀疏本地向量
// 方法第二个参数数组指定了非零元素的索引，而第三个参数数组则给定了非零元素值
scala> val sv1: Vector = Vectors.sparse(3, Array(0, 2), Array(2.0, 8.0))
sv1: org.apache.spark.mllib.linalg.Vector = (3,[0,2],[2.0,8.0])
// 另一种创建稀疏本地向量的方法
// 方法的第二个参数是一个序列，其中每个元素都是一个非零值的元组：(index,elem)
scala> val sv2: Vector = Vectors.sparse(3, Seq((0, 2.0), (2, 8.0)))
sv2: org.apache.spark.mllib.linalg.Vector = (3,[0,2],[2.0,8.0])

这里需要注意的是，Scala会默认引入scala.collection.immutable.Vector，我们要显式地引入org.apache.spark.mllib.linalg.Vector来使用MLlib提供的向量类型。

###二、标注点（Labeled Point）

标注点LabeledPoint是一种带有标签（Label/Response）的本地向量，它可以是稠密或者是稀疏的。在MLlib中，标注点在监督学习算法中被使用。由于标签是用双精度浮点型来存储的，故标注点类型在回归（Regression）和分类（Classification）问题上均可使用。例如，对于二分类问题，则正样本的标签为1，负样本的标签为0，而对于多类别的分类问题来说，标签则应是一个以0开始的索引序列:0, 1, 2 ...

标注点的实现类是org.apache.spark.mllib.regression.LabeledPoint，请注意它与前面介绍的本地向量不同，并不位于linalg包下，标注点的创建如下所示：

scala> import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.linalg.Vectors
scala> import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.mllib.regression.LabeledPoint
//创建一个标签为1.0（分类中可视为正样本）的稠密向量标注点
scala> val pos = LabeledPoint(1.0, Vectors.dense(2.0, 0.0, 8.0))
pos: org.apache.spark.mllib.regression.LabeledPoint = (1.0,[2.0,0.0,8.0])
//创建一个标签为0.0（分类中可视为负样本）的稀疏向量标注点
scala> val neg = LabeledPoint(0.0, Vectors.sparse(3, Array(0, 2), Array(2.0, 8.0)))
neg: org.apache.spark.mllib.regression.LabeledPoint = (0.0,(3,[0,2],[2.0,8.0]))

在实际的机器学习问题中，稀疏向量数据是非常常见的，MLlib提供了读取LIBSVM格式数据的支持，该格式被广泛用于LIBSVM、LIBLINEAR等机器学习库。在该格式下，每一个带标注的样本点由以下格式表示：

label index1:value1 index2:value2 index3:value3 ...

其中label是该样本点的标签值，一系列index:value对则代表了该样本向量中所有非零元素的索引和元素值。这里需要特别注意的是，index是以1开始并递增的。 MLlib在org.apache.spark.mllib.util.MLUtils工具类中提供了读取LIBSVM格式的方法loadLibSVMFile，其使用非常方便。

scala> import org.apache.spark.mllib.util.MLUtils
import org.apache.spark.mllib.util.MLUtils

// 用loadLibSVMFile方法读入LIBSVM格式数据
// sample_libsvm_data.txt为spark自带的一个示例，在以下地址可以找到：
// $SPARK_HOME$/data/mllib/sample_libsvm_data.txt
scala> val examples = MLUtils.loadLibSVMFile(sc, "/data/mllib/sample_libsvm_data.txt")
//返回的是组织成RDD的一系列LabeledPoint
examples: org.apache.spark.rdd.RDD[org.apache.spark.mllib.regression.LabeledPoint] = MapPartitionsRDD[6] at map at MLUtils.scala:108

这里，sc是Spark-shell自动建立的SparkContext。我们可以查看下加载进来的标注点的值：

scala> examples.collect().head
res7: org.apache.spark.mllib.regression.LabeledPoint = (0.0,(692,[127,128,129,130,131,154,155,156,157,158,159,181,182,183,184,185,186,187,188,189,207,208,209,210,211,212,213,214,215,216,217,235,236,237,238,239,240,241,242,243,244,245,262,263,264,265,266,267,268,269,270,271,272,273,289,290,291,292,293,294,295,296,297,300,301,302,316,317,318,319,320,321,328,329,330,343,344,345,346,347,348,349,356,357,358,371,372,373,374,384,385,386,399,400,401,412,413,414,426,427,428,429,440,441,442,454,455,456,457,466,467,468,469,470,482,483,484,493,494,495,496,497,510,511,512,520,521,522,523,538,539,540,547,548,549,550,566,567,568,569,570,571,572,573,574,575,576,577,578,594,595,596,597,598,599,600,601,602,603,604,622,623,624,625,626,627,628,629,630,651,652,653,654,655,656,657],[51.0,159.0,253.0,159.0,50...

这里，examples.collect()把rdd转换为了向量，并取第一个元素的值。每个标注点共有692个维，其中第127列对应的值是51.0，第128列对应的值是159.0，依此类推。

###三、本地矩阵（Local Matrix）

本地矩阵具有整型的行、列索引值和双精度浮点型的元素值，它存储在单机上。MLlib支持稠密矩阵DenseMatrix和稀疏矩阵Sparse Matrix两种本地矩阵，稠密矩阵将所有元素的值存储在一个列优先（Column-major）的双精度型数组中，而稀疏矩阵则将非零元素以列优先的CSC（Compressed Sparse Column）模式进行存储，关于CSC等稀疏矩阵存储方式的具体实现，可以参看Sparse Matrix Compression Formats一文。

本地矩阵的基类是org.apache.spark.mllib.linalg.Matrix，DenseMatrix和SparseMatrix均是它的实现类，和本地向量类似，MLlib也为本地矩阵提供了相应的工具类Matrices，调用工厂方法即可创建实例：

scala>import org.apache.spark.mllib.linalg.{Matrix, Matrices}
import org.apache.spark.mllib.linalg.{Matrix, Matrices}
// 创建一个3行2列的稠密矩阵[ [1.0,2.0], [3.0,4.0], [5.0,6.0] ]
// 请注意，这里的数组参数是列先序的！
scala> val dm: Matrix = Matrices.dense(3, 2, Array(1.0, 3.0, 5.0, 2.0, 4.0, 6.0))
dm: org.apache.spark.mllib.linalg.Matrix =
1.0  2.0
3.0  4.0
5.0  6.0

这里可以看出列优先的排列方式，即按照列的方式从数组中提取元素。也可以创建稀疏矩阵：

// 创建一个3行2列的稀疏矩阵[ [9.0,0.0], [0.0,8.0], [0.0,6.0]]
// 第一个数组参数表示列指针，即每一列元素的开始索引值
// 第二个数组参数表示行索引，即对应的元素是属于哪一行
// 第三个数组即是按列先序排列的所有非零元素，通过列指针和行索引即可判断每个元素所在的位置
scala> val sm: Matrix = Matrices.sparse(3, 2, Array(0, 1, 3), Array(0, 2, 1), Array(9, 6, 8))
sm: org.apache.spark.mllib.linalg.Matrix =
3 x 2 CSCMatrix
(0,0) 9.0
(2,1) 6.0
(1,1) 8.0

这里，创建一个3行2列的稀疏矩阵[ [9.0,0.0], [0.0,8.0], [0.0,6.0]]。Matrices.sparse的参数中，3表示行数，2表示列数。第1个数组参数表示列指针，即每一列元素的开始索引值， 第二个数组参数表示行索引，即对应的元素是属于哪一行；第三个数组即是按列先序排列的所有非零元素，通过列指针和行索引即可判断每个元素所在的位置。比如取每个数组的第2个元素为2,1,6，表示第2列第1行的元素值是6.0。