??MLlib是Spark提供提供機(jī)器學(xué)習(xí)的庫,專為在集群上并行運(yùn)行的情況而設(shè)計(jì)。
MLlib包含很多機(jī)器學(xué)習(xí)算法,可在Spark支持的所有編程語言中使用。
成都創(chuàng)新互聯(lián)是一家專業(yè)提供饒平企業(yè)網(wǎng)站建設(shè),專注與成都網(wǎng)站制作、做網(wǎng)站、H5開發(fā)、小程序制作等業(yè)務(wù)。10年已為饒平眾多企業(yè)、政府機(jī)構(gòu)等服務(wù)。創(chuàng)新互聯(lián)專業(yè)網(wǎng)絡(luò)公司優(yōu)惠進(jìn)行中。
??MLlib設(shè)計(jì)理念是將數(shù)據(jù)以RDD的形式表示,然后在分布式數(shù)據(jù)集上調(diào)用各種算法。其實(shí),MLlib就是RDD上一系列可供調(diào)用的函數(shù)的集合。
??MLlib包含一些特有的數(shù)據(jù)類型,位于org.apache.spark.mllib包(Java/Scala)或pyspark.mllib(Python)中。主要的幾個(gè)類有:
Vector
一個(gè)本地向量(Local Vector)。索引是從0開始的,并且是整型。而值為 Double 類型,存儲(chǔ)于單個(gè)機(jī)器內(nèi)。
MLlib既支持稠密向量也支持稀疏向量,前者表示向量的每一位都存儲(chǔ),后者只存儲(chǔ)非零位以節(jié)約空間。
向量可以通過mllib.linalg.Vectors類創(chuàng)建
//創(chuàng)建稠密向量
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]
//創(chuàng)建稀疏向量
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
在分類和回歸之類的監(jiān)督式學(xué)習(xí)(supervised learning)算法中使用。
LabeledPoint表示帶標(biāo)簽的數(shù)據(jù)點(diǎn),包括一個(gè)特征向量與一個(gè)標(biāo)簽(由一個(gè)浮點(diǎn)數(shù)表示)。
位于mllib.regression包中
// 首先需要引入標(biāo)簽點(diǎn)相關(guān)的類
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.regression.LabeledPoint
// 創(chuàng)建一個(gè)帶有正面標(biāo)簽和稠密特征向量的標(biāo)簽點(diǎn)。
val pos = LabeledPoint(1.0, Vectors.dense(1.0, 0.0, 3.0))
// 創(chuàng)建一個(gè)帶有負(fù)面標(biāo)簽和稀疏特征向量的標(biāo)簽點(diǎn)。
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
矩陣分為稠密矩陣和稀疏矩陣
稠密矩陣的實(shí)體值以列為主要次序的形式,存放于單個(gè) Double 型數(shù)組內(nèi)。系數(shù)矩陣的非零實(shí)體以列為主要次序的形式,存放于壓縮稀疏列(Compressed Sparse Column, CSC)中。例如,下面這個(gè)稠密矩陣就是存放在一維數(shù)組 [1.0, 3.0, 5.0, 2.0, 4.0, 6.0] 中,矩陣的大小為 (3, 2) 。
本地矩陣的基類是 Matrix 類,在 Spark 中有其兩種實(shí)現(xiàn),分別是 DenseMatrix 和 SparseMatrix 。官方文檔中推薦使用 已在 Matrices 類中實(shí)現(xiàn)的工廠方法來創(chuàng)建本地矩陣。需要注意的是,MLlib 中的本地矩陣是列主序的(column-major)
稠密矩陣
import org.apache.spark.mllib.linalg.{Matrix, Matrices}
// 創(chuàng)建稠密矩陣 ((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
用于產(chǎn)品推薦
表示用戶對(duì)一個(gè)產(chǎn)品的評(píng)分
各種Model類(模型)
每個(gè)Model都是訓(xùn)練算法的結(jié)果
??不論是在即時(shí)的探索中,還是在機(jī)器學(xué)習(xí)的數(shù)據(jù)理解中,基本的統(tǒng)計(jì)都是數(shù)據(jù)分析的重要部分。MLlib 通過mllib.stat.Statistics 類中的方法提供了幾種廣泛使用的統(tǒng)計(jì)函數(shù),這些函數(shù)可以直接在RDD 上使用。一些常用的函數(shù)如下所列。
??計(jì)算由向量組成的RDD 的匯總統(tǒng)計(jì),保存著向量集合中每列的最小值、最大值、平均值和方差。這可以用來在一次執(zhí)行中獲取豐富的統(tǒng)計(jì)信息。
?&esmp;計(jì)算由向量組成的RDD 中的列間的相關(guān)矩陣,使用皮爾森相關(guān)(Pearson correlation)或斯皮爾曼相關(guān)(Spearman correlation)中的一種(method 必須是pearson 或spearman中的一個(gè))。
??計(jì)算兩個(gè)由浮點(diǎn)值組成的RDD 的相關(guān)矩陣,使用皮爾森相關(guān)或斯皮爾曼相關(guān)中的一種(method 必須是pearson 或spearman 中的一個(gè))。
??計(jì)算由LabeledPoint 對(duì)象組成的RDD 中每個(gè)特征與標(biāo)簽的皮爾森獨(dú)立性測(cè)試
(Pearson’s independence test) 結(jié)果。返回一個(gè)ChiSqTestResult 對(duì)象, 其中有p 值、(p-value)、測(cè)試統(tǒng)計(jì)及每個(gè)特征的自由度。標(biāo)簽和特征值必須是分類的(即離散值)。
??下面舉個(gè)例子:使用三個(gè)學(xué)生的成績Vector來構(gòu)建所需的RDD Vector,這個(gè)矩陣?yán)锏拿總€(gè)Vector都代表一個(gè)學(xué)生在四門課程里的分?jǐn)?shù):
from pyspark.mllib.stat import Statistics
from pyspark.mllib.linalg import Vectors
//構(gòu)建RDD
basicTestRDD = sc.parallelize([Vectors.dense([60, 70, 80, 0]),
Vectors.dense([80, 50, 0, 90]),
Vectors.dense([60, 70, 80, 0])])
//查看summary里的成員,這個(gè)對(duì)象中包含了大量的統(tǒng)計(jì)內(nèi)容
>>> 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)
//創(chuàng)建一個(gè)HashingTF實(shí)例來把郵件文本映射為包含一個(gè)10000個(gè)特征的向量
val tf = new HashingTF(numFeatures = 10000)
//各郵件都被切分為單詞,每個(gè)單詞被映射為一個(gè)特征
val spamFeatures = spam.map { email => tf.transform(email.split(" ")) }
val normalFeatures = normal.map { email => tf.transform(email.split(" ")) }
//創(chuàng)建LabeledPoint數(shù)據(jù)集分別存放陽性(垃圾郵件)和陰性(正常郵件)的例子
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算法運(yùn)行邏輯回歸
val model = new LogisticRegressionWithSGD().run(trainingData)
//以陽性(垃圾郵件)和陰性(正常郵件)的例子分別進(jìn)行測(cè)試
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
# 讀取數(shù)據(jù)
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]
# 打印分割后的數(shù)據(jù)量
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))
#輸出結(jié)果,包含預(yù)測(cè)的數(shù)字結(jié)果和0/1結(jié)果:
for score, label in scoreAndLabels.collect():
print score, label
# 讀取數(shù)據(jù)文件,創(chuàng)建RDD
dataFile = "/opt/spark-1.6.1-bin-hadoop2.6/data/mllib/kmeans_data.txt"
lines = sc.textFile(dataFile)
# 創(chuàng)建Vector,將每行的數(shù)據(jù)用空格分隔后轉(zhuǎn)成浮點(diǎn)值返回numpy的array
data = lines.map(lambda line: np.array([float(x) for x in line.split(' ')]))
# 其中2是簇的個(gè)數(shù)
model = KMeans.train(data, 2)
print("Final centers: " + str(model.clusterCenters))
print("Total Cost: " + str(model.computeCost(data)))
忠于技術(shù),熱愛分享。歡迎關(guān)注公眾號(hào):java大數(shù)據(jù)編程,了解更多技術(shù)內(nèi)容。
網(wǎng)站題目:14.sparkmllib之快速入門
文章來源:http://bm7419.com/article46/pdhdeg.html
成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián),為您提供靜態(tài)網(wǎng)站、外貿(mào)網(wǎng)站建設(shè)、域名注冊(cè)、網(wǎng)站建設(shè)、搜索引擎優(yōu)化、企業(yè)網(wǎng)站制作
聲明:本網(wǎng)站發(fā)布的內(nèi)容(圖片、視頻和文字)以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主,如果涉及侵權(quán)請(qǐng)盡快告知,我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如需處理請(qǐng)聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載,或轉(zhuǎn)載時(shí)需注明來源: 創(chuàng)新互聯(lián)