Spark MLlib数据类型

时间:2023-03-09 18:46:40
Spark MLlib数据类型
    MLlib支持几种数据类型:本地向量(local vectors),和存储在一个简单机器中的矩阵(matrices),以及由一个或多个RDDs组成的分布式矩阵.
1,本地向量(Local Vector)
    一个本地向量是由从0开始的整型下标和double型值组成,存储在一个单机节点上.MLlib支持两种类型的本地向量:密集的和稀疏的.密集向量用一个double数组来存储值.而一个稀疏向量由两个并列的数组,下表和值组成.例如,一个向量(1.0, 0.0, 3.0)可以由密集的数组[1.0,0.0,3.0]表示,或者可以由(3,[0,2],[1.0,3.0])表示,其中3是指向量大小.后面表示,第0个元素是1.0,第2个元素是3.0.由此可看出,0.0在稀疏向量中可省略不写.
    本地向量的基本类型是Vector,并且有两个子类型:DenseVector,SparseVector.建议使用Vectors中的工厂方法来创建本地向量.
    import org.apache.spark.mllib.linalg.{Vector,Vectors}
    //创建一个密集向量(1.0,0.0,3.0)
    val dv:Vector = Vectors.dense(1.0,0.0,3.0)
    //通过设置非零元素的下标和值来创建一个稀疏向量(1.0,0.0,3.0)
    val sv1:Vector = Vectors.sparse(3,Array(0,2),Array(1.0,3.0))
    //通过另一种方式创建稀疏向量,采用下标,值对形式
    val sv2:Vector = Vectors.sparse(3,Seq((0,1.0),(2,3.0)))
注意:Scala默认imports scala.collection.immutable.Vector,所以一定要记得手动引入MLlib的Vector类

2,带标签的点(Labeled point)
    带标签的点实质上是一个本地向量(既可以是密集向量也可以是稀疏向量)和类标签(label/response).在MLlib中,带标签的点一般用在有监督学习算法中.用double数据存储一个标签,所以可以把带标签的点用在回归和分类算法(regression and classification)中.对于二分型分类算法,标签用0和1表示.对多分型分类算法,标签可以用从0开始的下标表示:0,1,2.....
    MLlib中带标签的点用LabeledPoint类表示
    import org.apache.spark.mllib.linglg.Vectors
    import org.apache.spark.mllib.regression.LabeledPoint
    //创建一个标签为1,数据为密集型向量构成的带标签点
    val pos = LabeledPoint(1.0, Vectors.dense(1.0,0.0,3.0))
    //创建一个标签为0,数据为稀疏型向量构成的带标签点
    val neg = LabeledPoint(0.0, Vectors.sparse(3,Array(0,2),Array(1.0,3.0)))

    (1)稀疏数据(sparse data)
    使用稀疏数据是很普遍的现象.MLlib支持读取以LIBSVM格式存储的训练数据,LIBSVM是LIBSVM和LIBLINEAR默认的格式.它是一种text格式的数据,里面的每一行代表一个含类标签的稀疏向量,格式如下:
    label index1:value1 index2:value2 .....
    LIBSVM中index是从1开始的,在加载完成后索引被转换成从0开始.
    用MLUtins.loadLibSVMFile从LIBSVM格式中读取数据
    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")
    SPARK_HOME/data/mllib/sample_libSVM_data.txt中数据格式如下
Spark MLlib数据类型
红色方框中的0和1代表标签类型

3,本地矩阵(Local matrix)
     本地矩阵由下标和double型的数值组成,其中下标是由整型数据表示行和列,保存在单个机器中。MLlib支持稀疏矩阵和密集矩阵,密集矩阵的值存在一个double数组中,一列列的进行存储。如下所示为一个密集矩阵
     Spark MLlib数据类型
表示一个size为(3,2)的密集矩阵,其值存在一个一维数组中,数组值为[1.0, 3.0, 5.0, 2.0, 4.0, 6.0]。稀疏矩阵的非零元素值都保存在压缩的稀疏列(Compressed Sparse Column,CSC)中,按列优先的顺序存储。
     本地矩阵基本类型为Matrix,有两个子类型:DenseMatrix和SparseMatrix。推荐使用Matrices的工厂方法来生成本地矩阵。必须牢记,MLlib中的本地矩阵数值是按列的顺序进行存储的。
     import org.apache.spark.mllib.linalg.{Matrix,Matrices}
     //创建一个密集矩阵((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))
     //创建一个稀疏矩阵((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))
注:稀疏矩阵解释,首先指定矩阵是3行2列,Array(0,1,3)是指,第0个非零元素为一列,第1,第二个非零元素为第二列.Array(0,2,1)是指,第一个非零元素在第0行,第二个非零元素在第2行,第三个非零元素在第1行.此处设计比较好,假设100个元素分两列,不需要把每个元素所在列都标出来,只需要记录3个数字即可.
Array(9,6,8)表示按顺序存储非零元素.

4,分布式矩阵(Distributed Matrix)
     一个分布式矩阵也是由下标和double型的值组成,不过分布式矩阵的下标不是Int型,而是long型,数据分布式保存在一个或多个RDD中。选择正确的格式来保存海量和分布式的矩阵是非常重要的。将分布式矩阵转换成不同的格式需要一个全局的shuffle(global shuffle),而全局shuffle的代价会非常高。到目前为止,已经实现了三种分布式矩阵。
     最基本的分布式矩阵是RowMatrix,它是一个行式的分布式矩阵,没有行索引。比如一系列特征向量的集合。RowMatrix由一个RDD代表所有的行,每一行是一个本地向量。假设一个RowMatrix的列数不是特别巨大,那么一个简单的本地向量能够与driver进行联系,并且数据可以在单个节点上保存或使用。IndexedRowMatrix与RowMatrix类似但是有行索引,行索引可以用来区分行并且进行连接等操作。CoordinateMatrix是一个以协同列表(coordinate list)格式存储数据的分布式矩阵,数据以RDD形式存储。
     注意:因为我们需要缓存矩阵的大小,所以分布式矩阵的RDDs格式是需要确定的,使用非确定RDDs的话会报错。
     (1)RowMatrix
     RowMatrix,它是一个行式的分布式矩阵,没有行索引。比如一系列特征向量的集合。RowMatrix由一个RDD代表所有的行,每一行是一个本地向量。因为每一行代表一个本地向量,所以它的列数被限制在Integer.max的范围内,在实际应用中不会太大。
     一个RowMatrix可以由一个RDD[Vector]的实例创建。因此我们可以计算统计信息或者进行分解。QR分解(QR decomposition)是A=QR,其中Q是一个矩阵,R是一个上三角矩阵。对sigular value decomposition(SVD)和principal component analysis(PCA),可以去参考降维的部分。
     import org.apache.spark.mllib.linalg.Vector
     import org.apache.spark.mllib.linalg.distributed.RowMatrix

     val rows:RDD[Vector] = ....//一个本地向量的RDD
     //从RDD[Vector]创建一个RowMatrix
     val mat: RowMatrix = new RowMatrix(rows)
     //获取RowMatrix的维度
     val m = mat.numRows()
     val n = mat.numCols()
     //QR降维
     val qrResult = mat.takkSkinnyQR(true)

     (2)IndexedRowMatrix
     IndexedRowMatrix与RowMatrix类似,但是它有行索引。由一个行索引RDD表示,索引每一行由一个long型行索引和一个本地向量组成。
     一个IndexedRowMatrix可以由RDD[IndexedRow]的实例来生成,IndexedRow是一个(Long, Vector)的封装。去掉 行索引,IndexedRowMatrix能够转换成RowMatrix。
     import org.apache.spark.mllib.linalg.distributed.{IndexedRow, IndexedRowMatrix,RowMatrix}
     val rows: RDD[IndexedRow] = ...//一个indexed rows的RDD
     //从RDD[IndexedRow]创建一个IndexedRowMatrix
     val mat:IndexedMatrix = new IndexedRowMatrix(rows)
     //获取维度
     val m = mat.numRows()
     val n = mat.numCols()
     //去掉行索引,转换成RowMatrix
     val rowMat:RowMatrix = mat.toRowMatrix()

     (3)CoordinateMatrix
     CoordinateMatrix是一个分布式矩阵,其实体集合是一个RDD,每一个是一个三元组(i:Long, j:Long, value:Double)。其中i是行索引,j是列索引,value是实体的值。当矩阵的维度很大并且是稀疏矩阵时,才使用CoordinateMatrix。
     一个CoordinateMatrix可以通过一个RDD[MatrixEntry]的实例来创建,MatrixEntry是一个(Long, Long, Double)的封装。CoordinateMatrix可以通过调用toIndexedRowMatrix转换成一个IndexedRowMatrix。CoordinateMatrix的其他降维方法暂时还不支持(Spark-1.5)。
     import org.apache.spark.mllib.linalg.distributed.{CoordinateMatrix,MatrixEntry}
     val entries:RDD[MatrixEntry] = ...//一个matrix entries的RDD
     //由RDD[MatrixEntry]创建一个CoordinateMatrix
     val mat:CoordinateMatrix = new CoordinateMatrix(entries)
     //获取矩阵的维度
     val m = mat.numRows()
     val n = mat.numCols()
     //调用toIndexedRowMatrix转换成IndexedRowMatrix,它的行都是稀疏向量
     vavl indexedRowMatrix = mat.toIndexedRowMatrix()

     (4)BlockMatrix
     一个BlockMatrix是一个分布式的矩阵,由一个MatrixBlocks的RDD组成。MatrixBlock是一个三元组((Int, Int), Matrix),其中(Int, Int)是block的索引,Matrix是一个维度为rowsPerBlock * colsPerBlock的矩阵子矩阵