spark mlib 机器学习系列之二：spark mlib 基本数据类型的使用

àì夳堔傛蜴生んèń 2022-05-24 07:20 191阅读 0赞

### spark mlib 基本数据类型 ###

Local Vector 本地向量集，向spark 提供一组可操作的数据集合  
Labeled point 向量标签，让用户分类不同的数据集合。  
Local Matrix 本地矩阵， 将数据集合以矩阵的形式存储在本地集合中。  
Distributed Matrix 分布式矩阵，将数据以矩阵形式存储在分布式计算机中。  
（分布式行矩阵，带标签分布式行矩阵，分布式坐标矩阵）

package mlib
    
    import org.apache.spark.ml.feature.LabeledPoint
    import org.apache.spark.ml.linalg.{Matrices, Vector, Vectors}
    import org.apache.spark.mllib.linalg.distributed._
    import org.apache.spark.mllib.util.MLUtils
    import org.apache.spark.sql.SparkSession
    
    object MlibDataTypeExample {
        def main(args: Array[String]): Unit = {
            // Local Vector 仅支持整型和float 数字， 密集型向量和稀疏型向量
            val vd : Vector = Vectors.dense(2, 0, 6)
            println(vd + ": " + vd(2))  // 结果[2.0,0.0,6.0]: 6.0
            val vs : Vector = Vectors.sparse(4, Array(0, 1, 2, 3), Array(9, 5, 2, 7))
            println(vs + ": " + vs(0))  // 结果(4,[0,1,2,3],[9.0,5.0,2.0,7.0]): 9.0
            val vdDemo : Vector = Vectors.dense('2', '0', 'h')
    
            // LabelPoint 的使用
            println("------------------LabeledPoint--------------------")
            val vd03 = Vectors.dense(3, 4, 5)
            val pos01 = LabeledPoint(1, vd03)
            println(pos01.label)   
            println(pos01.features)
            //1.0
            //[3.0,4.0,5.0]
    
            val vd04 = Vectors.sparse(3, Array(0, 1, 2), Array(2.3, 3.4, 5.6))
            val pos02 = LabeledPoint(2, vd04)
            println(pos02.label)
            println(pos02.features)
            //2.0
            //(3,[0,1,2],[2.3,3.4,5.6])
    
            // 本地矩阵用法
            // 个数需要匹配上，否则会报错
            val mx  = Matrices.dense(2,3, Array(1,2,3,4,5,6))
            println("-------------本地矩阵---------------")
            println(mx)
            //1.0 3.0 5.0 
            //2.0 4.0 6.0 
    
    
            val spark = SparkSession
                .builder
                .master("local[*]")
                .appName("UseFulRddOpts")
                .getOrCreate()
            val sc = spark.sparkContext
    
            println("-----------从数据库或者文件中加载数据---------")
            println("-----------文件中的数据的小标需要从1开始---------")
            // 从数据库或者外部数据集进行加载
            val mu = MLUtils.loadLibSVMFile(sc,
                "file:/D:/DevDir/Conclusions/SparkMlib/src/main/resources/a.txt")
            // a.txt
            //1 1:1 2:2 3:3
            //2 1:90 2:-9 3:9
            //1 1:3 2:8 3:4
            //2 1:1 2:7 3:2
            mu.foreach(println)
            //(1.0,(3,[0,1,2],[3.0,8.0,4.0]))
            //(1.0,(3,[0,1,2],[1.0,2.0,3.0]))
            //(2.0,(3,[0,1,2],[1.0,7.0,2.0]))
            //(2.0,(3,[0,1,2],[90.0,-9.0,9.0]))
    
    
    
            // 分布式行矩阵的用法
            val rdd = sc.textFile("file:/D:/DevDir/Conclusions/SparkMlib/src/main/resources/b.txt")
            // b.txt
            //1 2 3
            //4 5 6
            val rdd2 = rdd.map(_.split(" "))
            rdd2.foreach(a => {println(a(0) + " : " + a(1) + " : "  + a(2))})
            //1 : 2 : 3
            //4 : 5 : 6
    
            val rdd3 = rdd2.map(line => {
                val arr = Array(line(0).toDouble, line(1).toDouble, line(2).toDouble)
                org.apache.spark.mllib.linalg.Vectors.dense(arr)})
            val matrix = new RowMatrix(rdd3)
            println(matrix.numCols)   // 代表列 3
            println(matrix.numRows)    // 代表行 2
    
            // 带有行标签的分布式行行矩阵
            val rdd4 = rdd3.map(vd => new IndexedRow(vd.size, vd))
            val irm = new IndexedRowMatrix(rdd4)
            println(irm.getClass)
            irm.rows.foreach(println)
            // 结果
            //IndexedRow(3,[4.0,5.0,6.0])
            //IndexedRow(3,[1.0,2.0,3.0])
    
            // 分布式坐标矩阵
            val rdd5 = rdd2.map(line => {
                new MatrixEntry(line(0).toLong, line(1).toLong, line(2).toDouble)
            })
    
            val crm = new CoordinateMatrix(rdd5)
            crm.entries.foreach(println)
            //MatrixEntry(4,5,6.0)
            //MatrixEntry(1,2,3.0)
    
        }
    }