MLlib是Spark提供提供機器學習的庫,專爲在集羣上並行運行的狀況而設計。
MLlib包含不少機器學習算法,可在Spark支持的全部編程語言中使用。java
MLlib設計理念是將數據以RDD的形式表示,而後在分佈式數據集上調用各類算法。其實,MLlib就是RDD上一系列可供調用的函數的集合。python
MLlib包含一些特有的數據類型,位於org.apache.spark.mllib包(Java/Scala)或pyspark.mllib(Python)中。主要的幾個類有:git
Vector算法
一個本地向量(Local Vector)。索引是從0開始的,而且是整型。而值爲 Double 類型,存儲於單個機器內。apache
MLlib既支持稠密向量也支持稀疏向量,前者表示向量的每一位都存儲,後者只存儲非零位以節約空間。編程
向量能夠經過mllib.linalg.Vectors類建立數組
//建立稠密向量 scala> val denseVec1 = Vectors.dense(1.0,2.0,3.0) denseVec1: org.apache.spark.mllib.linalg.Vector = [1.0,2.0,3.0] scala> val denseVec2 = Vectors.dense(Array(1.0,2.0,3.0)) denseVec2: org.apache.spark.mllib.linalg.Vector = [1.0,2.0,3.0] //建立稀疏向量 scala> val sparseVec1 = Vectors.sparse(4,Array(0,2),Array(1.0,2.0)) sparseVec1: org.apache.spark.mllib.linalg.Vector = (4,[0,2],[1.0,2.0])
>>> from pyspark.mllib.linalg import Vectors >>> den = Vectors.dense([1.0,2.0,3.0]) >>> den DenseVector([1.0, 2.0, 3.0]) >>> spa = Vectors.sparse(4,[0,2],[1.0,2.0]) >>> spa SparseVector(4, {0: 1.0, 2: 2.0})
LabeledPoint網絡
在分類和迴歸之類的監督式學習(supervised learning)算法中使用。dom
LabeledPoint表示帶標籤的數據點,包括一個特徵向量與一個標籤(由一個浮點數表示)。機器學習
位於mllib.regression包中
// 首先須要引入標籤點相關的類 import org.apache.spark.mllib.linalg.Vectors import org.apache.spark.mllib.regression.LabeledPoint // 建立一個帶有正面標籤和稠密特徵向量的標籤點。 val pos = LabeledPoint(1.0, Vectors.dense(1.0, 0.0, 3.0)) // 建立一個帶有負面標籤和稀疏特徵向量的標籤點。 val neg = LabeledPoint(0.0, Vectors.sparse(3, Array(0, 2), Array(1.0, 3.0)))
>>> from pyspark.mllib.regression import LabeledPoint >>> from pyspark.mllib.linalg import Vectors >>> pos = LabeledPoint(1.0,Vectors.dense([1.0,2.0,3.0])) >>> neg = LabeledPoint(0.0,Vectors.dense([1.0,2.0,3.0]))
Matrix
矩陣分爲稠密矩陣和稀疏矩陣
稠密矩陣的實體值以列爲主要次序的形式,存放於單個 Double 型數組內。係數矩陣的非零實體以列爲主要次序的形式,存放於壓縮稀疏列(Compressed Sparse Column, CSC)中。例如,下面這個稠密矩陣就是存放在一維數組 [1.0, 3.0, 5.0, 2.0, 4.0, 6.0] 中,矩陣的大小爲 (3, 2) 。
本地矩陣的基類是 Matrix 類,在 Spark 中有其兩種實現,分別是 DenseMatrix 和 SparseMatrix 。官方文檔中推薦使用 已在 Matrices 類中實現的工廠方法來建立本地矩陣。須要注意的是,MLlib 中的本地矩陣是列主序的(column-major)
稠密矩陣
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))
scala> val sparseMatrix= Matrices.sparse(3, 3, Array(0, 2, 3, 6), Array(0, 2, 1, 0, 1, 2), Array(1.0, 2.0, 3.0,4.0,5.0,6.0)) sparseMatrix: org.apache.spark.mllib.linalg.Matrix = 3 x 3 CSCMatrix (0,0) 1.0 (2,0) 2.0 (1,1) 3.0 (0,2) 4.0 (1,2) 5.0 (2,2) 6.0
Rating
用於產品推薦
表示用戶對一個產品的評分
各類Model類(模型)
每一個Model都是訓練算法的結果
不管是在即時的探索中,仍是在機器學習的數據理解中,基本的統計都是數據分析的重要部分。MLlib 經過mllib.stat.Statistics 類中的方法提供了幾種普遍使用的統計函數,這些函數能夠直接在RDD 上使用。一些經常使用的函數以下所列。
計算由向量組成的RDD 的彙總統計,保存着向量集合中每列的最小值、最大值、平均值和方差。這能夠用來在一次執行中獲取豐富的統計信息。
&esmp;計算由向量組成的RDD 中的列間的相關矩陣,使用皮爾森相關(Pearson correlation)或斯皮爾曼相關(Spearman correlation)中的一種(method 必須是pearson 或spearman中的一個)。
計算兩個由浮點值組成的RDD 的相關矩陣,使用皮爾森相關或斯皮爾曼相關中的一種(method 必須是pearson 或spearman 中的一個)。
計算由LabeledPoint 對象組成的RDD 中每一個特徵與標籤的皮爾森獨立性測試
(Pearson’s independence test) 結果。返回一個ChiSqTestResult 對象, 其中有p 值、(p-value)、測試統計及每一個特徵的自由度。標籤和特徵值必須是分類的(即離散值)。
下面舉個例子:使用三個學生的成績Vector來構建所需的RDD Vector,這個矩陣裏的每一個Vector都表明一個學生在四門課程裏的分數:
from pyspark.mllib.stat import Statistics from pyspark.mllib.linalg import Vectors //構建RDD basicTestRDD = sc.parallelize([Vectors.dense([60, 70, 80, 0]), Vectors.dense([80, 50, 0, 90]), Vectors.dense([60, 70, 80, 0])]) //查看summary裏的成員,這個對象中包含了大量的統計內容 >>> print summary.mean() [ 66.66666667 63.33333333 53.33333333 30. ] >>> print summary.variance() [ 133.33333333 133.33333333 2133.33333333 2700. ] >>> print summary.numNonzeros() [ 3. 3. 2. 1.]
import org.apache.spark.mllib.linalg.{Vector, Vectors} import org.apache.spark.rdd.RDD val array1: Array[Double] = Array[Double](60, 70, 80, 0) val array2: Array[Double] = Array[Double](80, 50, 0, 90) val array3: Array[Double] = Array[Double](60, 70, 80, 0) val denseArray1 = Vectors.dense(array1) val denseArray2 = Vectors.dense(array2) val denseArray3 = Vectors.dense(array3) val seqDenseArray: Seq[Vector] = Seq(denseArray1, denseArray2, denseArray3) val basicTestRDD: RDD[Vector] = sc.parallelize[Vector](seqDenseArray) val summary: MultivariateStatisticalSummary = Statistics.colStats(basicTestRDD)
主成分分析(PCA)
def testLogisticRegressionWithSGD = { val spam = sc.textFile("src/main/resources/mllib/spam.txt", 1) val normal = sc.textFile("src/main/resources/mllib/normal.txt", 1) //建立一個HashingTF實例來把郵件文本映射爲包含一個10000個特徵的向量 val tf = new HashingTF(numFeatures = 10000) //各郵件都被切分爲單詞,每一個單詞被映射爲一個特徵 val spamFeatures = spam.map { email => tf.transform(email.split(" ")) } val normalFeatures = normal.map { email => tf.transform(email.split(" ")) } //建立LabeledPoint數據集分別存放陽性(垃圾郵件)和陰性(正常郵件)的例子 val positiveExamples = spamFeatures.map { features => LabeledPoint(1, features) } val negativeExamples = normalFeatures.map { features => LabeledPoint(0, features) } val trainingData = positiveExamples.union(negativeExamples) trainingData.cache() println(trainingData.toDebugString) //使用SGD算法運行邏輯迴歸 val model = new LogisticRegressionWithSGD().run(trainingData) //以陽性(垃圾郵件)和陰性(正常郵件)的例子分別進行測試 val posTest = tf.transform("O M G get cheap stuff by sending money to .".split(" ")) val negTest = tf.transform("hello, i started studying Spark ".split(" ")) println(s"prediction for positive tset example: ${model.predict(posTest)}") println(s"prediction for negitive tset example: ${model.predict(negTest)}") Thread.sleep(Int.MaxValue) }
# 加載模塊 from pyspark.mllib.util import MLUtils from pyspark.mllib.classification import SVMWithSGD # 讀取數據 dataFile = '/opt/spark-1.6.1-bin-hadoop2.6/data/mllib/sample_libsvm_data.txt' data = MLUtils.loadLibSVMFile(sc, dataFile) splits = data.randomSplit([0.8, 0.2], seed = 9L) training = splits[0].cache() test = splits[1] # 打印分割後的數據量 print "TrainingCount:[%d]" % training.count(); print "TestingCount:[%d]" % test.count(); model = SVMWithSGD.train(training, 100) scoreAndLabels = test.map(lambda point : (model.predict(point.features), point.label)) #輸出結果,包含預測的數字結果和0/1結果: for score, label in scoreAndLabels.collect(): print score, label
# 讀取數據文件,建立RDD dataFile = "/opt/spark-1.6.1-bin-hadoop2.6/data/mllib/kmeans_data.txt" lines = sc.textFile(dataFile) # 建立Vector,將每行的數據用空格分隔後轉成浮點值返回numpy的array data = lines.map(lambda line: np.array([float(x) for x in line.split(' ')])) # 其中2是簇的個數 model = KMeans.train(data, 2) print("Final centers: " + str(model.clusterCenters)) print("Total Cost: " + str(model.computeCost(data)))
忠於技術,熱愛分享。歡迎關注公衆號:java大數據編程,瞭解更多技術內容。