Spark MLlib 之 Vector向量深入淺出
Spark MLlib裏面提供了幾種基本的數據類型,雖然大部分在調包的時候用不到,但是在自己寫算法的時候,還是很需要了解的。MLlib支持單機版本的local vectors向量和martix矩陣,也支持集群版本的matrix矩陣。他們背後使用的都是ScalaNLP中的Breeze。
更多內容參考我的大數據學習之路
Local Vector
local vector是一種索引是0開始的整數、內容為double類型,存儲在單機上的向量。MLlib支持兩種矩陣,dense密集型和sparse稀疏型。一個dense類型的向量背後其實就是一個數組,而sparse向量背後則是兩個並行數組——索引數組和值數組。比如向量(1.0, 0.0, 3.0)既可以用密集型向量表示為[1.0, 0.0, 3.0],也可以用稀疏型向量表示為(3, [0,2],[1.0,3.0]),其中3是數組的大小。
接口為Vector,看源碼可以看到它是用sealed修飾的,在scala裏面這種關鍵字修飾的trait在進行match的時候必須把所有的情況都列出來,不然會報錯。相當於強制你考慮向量的時候,必須考慮它是dense型的,還是sparse型的。
sealed trait Vector extends Serializable { def size: Int // 向量的大小 def toArray: Array[Double] //轉換成普通的數組 override def equals(other: Any): Boolean = { // 定義比較方法——感慨,原來這麽優秀的框架背後也用窮舉 other match { case v2: Vector => if (this.size != v2.size) return false (this, v2) match { case (s1: SparseVector, s2: SparseVector) => Vectors.equals(s1.indices, s1.values, s2.indices, s2.values) case (s1: SparseVector, d1: DenseVector) => Vectors.equals(s1.indices, s1.values, 0 until d1.size, d1.values) case (d1: DenseVector, s1: SparseVector) => Vectors.equals(0 until d1.size, d1.values, s1.indices, s1.values) case (_, _) => util.Arrays.equals(this.toArray, v2.toArray) } case _ => false } } override def hashCode(): Int = { //好好領略hashcode的魅力 var result: Int = 31 + size var nnz = 0 this.foreachActive { (index, value) => if (nnz < Vectors.MAX_HASH_NNZ) { if (value != 0) { result = 31 * result + index val bits = java.lang.Double.doubleToLongBits(value) result = 31 * result + (bits ^ (bits >>> 32)).toInt nnz += 1 } } else { return result } } result } // 這裏面的BV其實是breeze裏面的vector,import breeze.linalg.{DenseVector => BDV, SparseVector => BSV, Vector => BV} // 也就是說,mllib裏面的vector其實就是對breeze裏面的vector封裝了一層而已 private[spark] def asBreeze: BV[Double] def apply(i: Int): Double = asBreeze(i) def copy: Vector = { throw new NotImplementedError(s"copy is not implemented for ${this.getClass}.") } def foreachActive(f: (Int, Double) => Unit): Unit def numActives: Int def numNonzeros: Int //零的個數 def toSparse: SparseVector def toDense: DenseVector = new DenseVector(this.toArray) //創建Dense向量還真是簡單啊 def compressed: Vector = { val nnz = numNonzeros // A dense vector needs 8 * size + 8 bytes, while a sparse vector needs 12 * nnz + 20 bytes. if (1.5 * (nnz + 1.0) < size) { toSparse } else { toDense } } def argmax: Int //返回裏面的最大值 }
Vector有兩種實現方式——DenseVector,和SparseVector。
import org.apache.spark.ml.linalg.{Vector,Vectors} import org.apache.spark.sql.SparkSession object DataTypes { def main(args: Array[String]): Unit = { val spark = SparkSession.builder().master("local[*]").appName("tf-idf").getOrCreate() spark.sparkContext.setLogLevel("WARN") // 創建dense vector val dv: Vector = Vectors.dense(1.0, 0.0, 3.0) // 創建sparse vector 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))) } }
其中sparse vector有兩種創建方式,第一種是傳入三個參數:向量大小、索引數組、索引數組對應的值數組;第二種方式是傳入兩個參數:向量大小、由索引和值組成的鍵值對數組。
另外這個Vectors不僅僅有創建dense和sparse的方法,還有幾個有用的功能,比如norm範數和sqdist距離。
val norm1Vec = Vectors.dense(1.0,-1.0,2.0)
// 第一範數,就是絕對值相加
println(Vectors.norm(norm1Vec,1)) // 4.0
// 第二番薯,就是平方和開根號
println(Vectors.norm(norm1Vec,2)) // 2.449489742783178
// 無限範數
println(Vectors.norm(norm1Vec,1000)) //2.0
val sq1 = Vectors.dense(1.0, 2.0, 3.0)
val sq2 = Vectors.dense(2.0, 4.0, 6.0)
println(Vectors.sqdist(sq1, sq2)) // (2-1)^2 + (4-2)^2 + (6-3)^2 = 14
通過上面簡單的一個Vector,還是能學到不少東西的。
比如sealed關鍵字的使用,以及工廠方法:
object xxxFactory{
def x1: XXX
def x2: XXX
...
}
trait XXX {}
object X1 extends XXX {}
object X2 extends XXX {}
Labeled Point 有標簽的向量
這種labeled point其實內部也是一個vector,可能是dense也可能是sparse,不過多了一個標簽列。在ML裏面,labeled point通常用於有監督算法。這個label是double類型的,這樣既可以用於回歸算法,也可以用於分類。在二分類中,Label不是0就是1;在多分類中label可能從0開始,1,2,3,4....
使用的時候很簡單,直接new就可以了:
// 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)))
一般在準備訓練集數據的時候,數據都是稀疏型的。MMLib支持在SVM和Liner線性回歸中直接讀取訓練數據,但是需要滿足下面的格式:
label index1:value1 index2:value2 ...
比如:
val examples: RDD[LabeledPoint] = MLUtils.loadLibSVMFile(sc, "data/mllib/sample_libsvm_data.txt")
Spark MLlib 之 Vector向量深入淺出