第7章 Spark MLlib機(jī)器學(xué)習(xí)

SparkMLlib機(jī)器學(xué)習(xí)1 MLlib機(jī)器學(xué)習(xí)庫(kù)2 MLlib基本數(shù)據(jù)類(lèi)型3

機(jī)器學(xué)習(xí)流水線4

基本統(tǒng)計(jì)5

特征提取、轉(zhuǎn)換和選擇6分類(lèi)算法7回歸算法8聚類(lèi)算法9協(xié)同過(guò)濾推薦算法1MLlib機(jī)器學(xué)習(xí)庫(kù)MLlib（MachineLearningLibrary，機(jī)器學(xué)習(xí)庫(kù)）是Spark的機(jī)器學(xué)習(xí)庫(kù)，旨在簡(jiǎn)化機(jī)器學(xué)習(xí)的工程實(shí)踐工作。MLlib由一些通用的學(xué)習(xí)算法和工具組成，學(xué)習(xí)算法包括分類(lèi)、回歸、聚類(lèi)、協(xié)同過(guò)濾、降維等，可在Spark支持的所有編程語(yǔ)言中使用。MLlib設(shè)計(jì)理念是將數(shù)據(jù)以RDD的形式表示，然后在分布式數(shù)據(jù)集上調(diào)用各種算法。實(shí)際上，MLlib就是RDD上一系列可供調(diào)用的函數(shù)的集合。MLlib庫(kù)目前分為兩個(gè)代碼包：一個(gè)是spark.mllib，包含基于RDD的API，spark.mllib中的算法接口是基于RDD的；另一個(gè)是spark.ml，提供了基于DataFrame的高層次API，spark.ml中的算法接口是基于DataFrame的。Spark未來(lái)會(huì)主要支持ml。

SparkMLlib機(jī)器學(xué)習(xí)1 MLlib機(jī)器學(xué)習(xí)庫(kù)2 MLlib基本數(shù)據(jù)類(lèi)型3

機(jī)器學(xué)習(xí)流水線4

基本統(tǒng)計(jì)5

特征提取、轉(zhuǎn)換和選擇6分類(lèi)算法7回歸算法8聚類(lèi)算法9協(xié)同過(guò)濾推薦算法2MLlib基本數(shù)據(jù)類(lèi)型MLlib提供了一序列基本數(shù)據(jù)類(lèi)型以支持底層的機(jī)器學(xué)習(xí)算法，具體包括：本地向量（Localvector）、帶標(biāo)簽的點(diǎn)（Labeledpoint）、本地矩陣（Localmatrix）等。本地向量（LocalVector）本地向量（LocalVector）由從0開(kāi)始的整數(shù)類(lèi)型索引和double類(lèi)型的值組成。MLlib支持兩種類(lèi)型的本地向量：稠密型向量DenseVector和稀疏型向量SparseVector。稠密向量使用一個(gè)雙精度浮點(diǎn)型數(shù)組來(lái)表示向量中的每個(gè)元素，而一個(gè)稀疏型向量是由一個(gè)整型索引數(shù)組、一個(gè)雙精度浮點(diǎn)型數(shù)組來(lái)分別表示非零元素在向量中索引、向量中的非零元素。例如，向量(1.0,0.0,3.0)的稠密向量表示是[1.0,0.0,3.0]；稀疏向量表示是(3,[0,2],[1.0,3.0])，元組的第一個(gè)值3是向量的長(zhǎng)度，[0,2]是向量的非零元素的索引所組成的數(shù)組，[1.0,3.0]是向量中的非零元素組成的數(shù)組。所有本地向量的基類(lèi)都是pyspark.ml.linalg.Vectors，DenseVector和SparseVector分別是它的兩個(gè)實(shí)現(xiàn)類(lèi)2MLlib基本數(shù)據(jù)類(lèi)型1.稠密本地向量>>>frompyspark.ml.linalgimportVectors#導(dǎo)入Vectors類(lèi)>>>vector=Vectors.dense(1.0,0.0,3.0)#創(chuàng)建一個(gè)稠密本地向量>>>print(vector)#輸出vector[1.0,0.0,3.0]>>>vector.size#獲取向量的長(zhǎng)度3也可以采用如下方式創(chuàng)建稠密本地向量。>>>v=Vectors.dense([1.0,2.0])>>>u=Vectors.dense([3.0,4.0])>>>v+u#稠密本地向量之間可以進(jìn)行加減乘除DenseVector([4.0,6.0])>>>vector.toArray()#向量轉(zhuǎn)成數(shù)組array([1.,0.,3.])2MLlib基本數(shù)據(jù)類(lèi)型2.稀疏本地向量#創(chuàng)建一個(gè)稀疏本地向量，方法的第二個(gè)元素指定了非零元素的索引#方法的第三個(gè)參數(shù)是索引對(duì)應(yīng)的非零元素>>>sparseVector1=Vectors.sparse(3,[0,2],[1.0,3.0])#另一種創(chuàng)建稀疏本地向量的方法，方法的第二個(gè)參數(shù)是一個(gè)列表//其中的每個(gè)元素都是由向量中非零元素索引及其對(duì)應(yīng)的元素所組成的二元組>>>sparseVector2=Vectors.sparse(3,[(0,1.0),(2,3.0)])>>>print(sparseVector2)#輸出sparseVector2(3,[0,2],[1.0,3.0])還可以采用如下方式創(chuàng)建稀疏本地向量。>>>sparseVector3=Vectors.sparse(3,{0:1,2:3})>>>print(sparseVector3)(3,[0,2],[1.0,3.0])2MLlib基本數(shù)據(jù)類(lèi)型3.帶標(biāo)簽的點(diǎn)帶標(biāo)簽的點(diǎn)（LabeledPoint）是一種帶有標(biāo)簽（Label）的本地向量，向量可以是稠密的，也可以是稀疏的。下面給出創(chuàng)建帶標(biāo)簽的點(diǎn)的實(shí)例。>>>frompyspark.mllib.linalgimportSparseVector#導(dǎo)入SparseVector類(lèi)>>>frompyspark.mllib.regressionimportLabeledPoint#導(dǎo)入LabeledPoint類(lèi)#創(chuàng)建一個(gè)標(biāo)簽為1.0的帶標(biāo)簽的點(diǎn)，其向量為稠密本地向量>>>pos=LabeledPoint(1.0,[1.0,0.0,3.0])>>>print(pos)#輸出pos(1.0,[1.0,0.0,3.0])#創(chuàng)建一個(gè)標(biāo)簽為0.0的帶標(biāo)簽的點(diǎn)，其特征向量為稀疏本地向量>>>neg=LabeledPoint(0.0,SparseVector(3,[0,2],[1.0,3.0]))>>>print(neg)#輸出neg(0.0,(3,[0,2],[1.0,3.0]))2MLlib基本數(shù)據(jù)類(lèi)型4.本地矩陣本地矩陣由整數(shù)型的行和列的索引以及雙精度浮點(diǎn)型的元素值組成，MLlib支持稠密矩陣DenseMatrix和稀疏矩陣SparseMatrix兩種本地矩陣。稠密矩陣將所有元素的值存儲(chǔ)在一個(gè)列優(yōu)先的雙精度型數(shù)組中，而稀疏矩陣則將非零元素按列優(yōu)先的次序以以CSC（CompressedSparseColumn）格式進(jìn)行存儲(chǔ)。本地矩陣的基類(lèi)是pyspark.ml.linalg.Matrix，DenseMatrix和SparseMatrix均是它的實(shí)現(xiàn)類(lèi)，和本地向量類(lèi)似，MLlib也為本地矩陣提供了相應(yīng)的工具類(lèi)Matrices，調(diào)用它的工廠方法即可創(chuàng)建本地矩陣實(shí)例。>>>frompyspark.ml.linalgimportMatrices#創(chuàng)建3行2列的稠密矩陣>>>Matrix=Matrices.dense(3,2,[1.0,3.0,5.0,2.0,4.0,6.0])>>>print(Matrix)DenseMatrix([[1.,2.],[3.,4.],[5.,6.]])2MLlib基本數(shù)據(jù)類(lèi)型4.本地矩陣

#創(chuàng)建一個(gè)4行3列的稀疏矩陣>>>sparseMatrix=Matrices.sparse(4,3,[0,2,4,6],[0,1,1,2,2,3],[1,2,3,4,5,6])>>>print(sparseMatrix)4X3CSCMatrix(0,0)1.0(1,0)2.0(1,1)3.0。。。。第1個(gè)列表參數(shù)[0,2,4,6]表示要?jiǎng)?chuàng)建的矩陣的索引為0的列有2-0個(gè)非零元素，索引為1的列有4-2個(gè)非零元素，索引為2的列有6-4個(gè)非零元素，6表示矩陣的非零元素個(gè)數(shù)。第2個(gè)列表參數(shù)[0,1,1,2,2,3]表示列優(yōu)先排序的非0元素的行索引，其列表長(zhǎng)度等于非零元素的個(gè)數(shù)；第3個(gè)列表參數(shù)[1,2,3,4,5,6]是按列優(yōu)先排序的所有非零元素。

SparkMLlib機(jī)器學(xué)習(xí)1 MLlib機(jī)器學(xué)習(xí)庫(kù)2 MLlib基本數(shù)據(jù)類(lèi)型3 機(jī)器學(xué)習(xí)流水線4

基本統(tǒng)計(jì)5

特征提取、轉(zhuǎn)換和選擇6分類(lèi)算法7回歸算法8聚類(lèi)算法9協(xié)同過(guò)濾推薦算法3機(jī)器學(xué)習(xí)流水線一個(gè)典型的機(jī)器學(xué)習(xí)過(guò)程通常包含數(shù)據(jù)加載、數(shù)據(jù)預(yù)處理、特征提取、模型訓(xùn)練、模型評(píng)估、數(shù)據(jù)預(yù)測(cè)等步驟。Transformerpyspark.ml的三個(gè)主要的抽象類(lèi)：轉(zhuǎn)換器Transformer、評(píng)估器Estimator、流水線Pipeline。轉(zhuǎn)換器Transformer將一個(gè)DataFrame轉(zhuǎn)換為另一個(gè)DataFrame。Transformer的方法transform()通過(guò)附加一個(gè)或多個(gè)列將一個(gè)DataFrame轉(zhuǎn)換為另一個(gè)DataFrame。>>>frompyspark.ml.linalgimportVectors>>>frompyspark.ml.featureimportPCA>>>data=[(Vectors.sparse(5,[(1,2.0),(3,4.0)]),),

(Vectors.dense([2.0,0.0,4.0,6.0,7.0]),),

(Vectors.dense([4.0,0.0,5.0,6.0,7.0]),)]3機(jī)器學(xué)習(xí)流水線Transformer>>>df=spark.createDataFrame(data,["features"])>>>pca=PCA(k=3,inputCol="features",outputCol="pcaFeatures")>>>model=pca.fit(df)#訓(xùn)練模型>>>result=model.transform(df).select("pcaFeatures")#轉(zhuǎn)換數(shù)據(jù)>>>result.show(truncate=False)#查看數(shù)據(jù)+---------------------------------------------------------------+|pcaFeatures|+---------------------------------------------------------------+|[-0.42573613396564325,0.2607104294741789,-0.007815802912680248]||[-9.127205618459051,1.3940368654272142,-0.007815802912681136]||[-10.294958452424574,-0.5128856077649964,-0.007815802912681136]|+---------------------------------------------------------------+3機(jī)器學(xué)習(xí)流水線Estimator評(píng)估器Estimator是學(xué)習(xí)算法在訓(xùn)練數(shù)據(jù)上的訓(xùn)練的概念抽象?？梢酝ㄋ桌斫鉃樗惴āＴ赑ipeline里通常是被用來(lái)操作DataFrame數(shù)據(jù)并生成一個(gè)Transformer。從技術(shù)上講，Estimator實(shí)現(xiàn)了一個(gè)方法fit()，它接受一個(gè)DataFrame并產(chǎn)生一個(gè)轉(zhuǎn)換器。比如，一個(gè)主成分分析算法PCA就是一個(gè)Estimator，它可以調(diào)用fit()，通過(guò)訓(xùn)練特征數(shù)據(jù)而得到一個(gè)主成分分析轉(zhuǎn)換器。Pipeline流水線Pipeline將多個(gè)工作流階段（轉(zhuǎn)換器和估計(jì)器）連接在一起，形成機(jī)器學(xué)習(xí)的工作流，并獲得結(jié)果輸出。要構(gòu)建一個(gè)Pipeline流水線，首先需要定義Pipeline中的各個(gè)流水線階段PipelineStage（包括轉(zhuǎn)換器和評(píng)估器），有了這些處理特定問(wèn)題的轉(zhuǎn)換器和評(píng)估器，就可以按照具體的處理邏輯有序地組織PipelineStages并創(chuàng)建一個(gè)Pipeline。3機(jī)器學(xué)習(xí)流水線Pipeline>>>pipeline=Pipeline(stages=[stage1,stage2,stage3])然后就可以把訓(xùn)練數(shù)據(jù)集作為輸入?yún)?shù)，調(diào)用Pipeline實(shí)例的fit方法來(lái)開(kāi)始以流的方式來(lái)處理源訓(xùn)練數(shù)據(jù)。這個(gè)調(diào)用會(huì)返回一個(gè)PipelineModel類(lèi)實(shí)例，它是一個(gè)Transformer，進(jìn)而被用來(lái)預(yù)測(cè)測(cè)試數(shù)據(jù)的標(biāo)簽。下面以邏輯回歸為例，構(gòu)建一個(gè)典型的機(jī)器學(xué)習(xí)過(guò)程，具體介紹流水線是如何應(yīng)用的。（1）引入要包含的庫(kù)并構(gòu)建訓(xùn)練數(shù)據(jù)集。>>>frompyspark.mlimportPipeline>>>frompyspark.ml.classificationimportLogisticRegression>>>frompyspark.ml.featureimportHashingTF,Tokenizer#創(chuàng)建由三元組列表(id,text,label)組成傳遞訓(xùn)練文檔>>>training=spark.createDataFrame([(0,"RobertDeNiroplaysthemost.",0.0),(2,"Ifyoulike",1.0),(3,"AlthoughIdidn't.",1.0)],["id","text","label"])3機(jī)器學(xué)習(xí)流水線Pipeline（2）定義Pipeline中的各個(gè)流水線階段PipelineStage，包括轉(zhuǎn)換器和評(píng)估器，具體地，包含tokenizer,hashingTF和LR。>>>tokenizer=Tokenizer(inputCol="text",outputCol="words")#輸入列名，輸出列名#單詞轉(zhuǎn)化為特征向量>>>hashingTF=HashingTF(inputCol=tokenizer.getOutputCol(),outputCol="features")#創(chuàng)建模型，設(shè)置最大迭代次數(shù)，學(xué)習(xí)率>>>LR=LogisticRegression(maxIter=10,regParam=0.001)（3）按照具體的處理邏輯有序地組織PipelineStages，并創(chuàng)建一個(gè)Pipeline。>>>pipeline=Pipeline(stages=[tokenizer,hashingTF,LR])（4）現(xiàn)在構(gòu)建的pipeline本質(zhì)上是一個(gè)Estimator，在他調(diào)用fit()方法運(yùn)行之后，它將產(chǎn)生一個(gè)PipelineModel，它是一個(gè)Transformer。>>>model=pipeline.fit(training)3機(jī)器學(xué)習(xí)流水線Pipeline（5）構(gòu)建測(cè)試數(shù)據(jù)。>>>test=spark.createDataFrame([(4,"Anobviousvanitypresswon'teither."),(5,"Thisisoneofmythreebreakingclimax.")],["id","text"])（6）調(diào)用之前訓(xùn)練好的PipelineModel的transform()方法，讓測(cè)試數(shù)據(jù)按順序通過(guò)擬合的流水線，生成預(yù)測(cè)結(jié)果。>>>prediction=model.transform(test)>>>selected=prediction.select("id","probability","prediction")>>>forrowinselected.collect():...rid,prob,prediction=row...print("%d-->prob=%s,prediction=%f"%(rid,str(prob),prediction))...4-->prob=[0.2044020986901636,0.7955979013098364],prediction=1.0000005-->prob=[0.5339925451618571,0.4660074548381429],prediction=0.000000

SparkMLlib機(jī)器學(xué)習(xí)1 MLlib機(jī)器學(xué)習(xí)庫(kù)2 MLlib基本數(shù)據(jù)類(lèi)型3 機(jī)器學(xué)習(xí)流水線4 基本統(tǒng)計(jì)5

特征提取、轉(zhuǎn)換和選擇6分類(lèi)算法7回歸算法8聚類(lèi)算法9協(xié)同過(guò)濾推薦算法4基本統(tǒng)計(jì)匯總統(tǒng)計(jì)給定一個(gè)數(shù)據(jù)集，數(shù)據(jù)分析人員一般會(huì)先觀察一下數(shù)據(jù)集的整體情況，稱(chēng)之為匯總統(tǒng)計(jì)或者概要性統(tǒng)計(jì)。匯總統(tǒng)計(jì)的基本統(tǒng)計(jì)量包括描述數(shù)據(jù)集中趨勢(shì)的統(tǒng)計(jì)值（平均數(shù)、中位數(shù)和眾數(shù)）、描述數(shù)據(jù)離中趨勢(shì)的統(tǒng)計(jì)量（極差、四分位數(shù)、平均差、方差、標(biāo)準(zhǔn)差和變異系數(shù)）和描述數(shù)據(jù)分布狀況的統(tǒng)計(jì)量（偏態(tài)系數(shù)）。有了這些基本統(tǒng)計(jì)量，數(shù)據(jù)分析人員就掌握了數(shù)據(jù)的基本特征，進(jìn)而基本確定對(duì)數(shù)據(jù)做進(jìn)一步分析的方向。PySpark用Summarizer來(lái)進(jìn)行匯總統(tǒng)計(jì)，Summarizer位于pyspark.ml.stat包下，主要用來(lái)對(duì)DataFrame進(jìn)行統(tǒng)計(jì)，比如最大值、最小值、方差、均值、非零值等等。frompyspark.ml.statimportSummarizerfrompyspark.sqlimportRowfrompyspark.ml.linalgimportVectorsfrompysparkimportSparkConf,SparkContext4基本統(tǒng)計(jì)匯總統(tǒng)計(jì)frompyspark.sqlimportSparkSessionspark=SparkSession.builder.getOrCreate()#生成DataFrame，先利用parallelize()方法生成帶權(quán)重的RDD，再用toDF()方法將其轉(zhuǎn)為DataFramedf=spark.sparkContext.parallelize([Row(weight=0.8,features=Vectors.dense(1.0,1.0,1.0)),Row(weight=0.5,features=Vectors.dense(1.0,2.0,3.0))]).toDF()#創(chuàng)建Summarizer模型，有多個(gè)統(tǒng)計(jì)量時(shí)利用Summarizer.metrics(統(tǒng)計(jì)量1,(統(tǒng)計(jì)量1,......)summarizer=Summarizer.metrics("mean","count")#計(jì)算統(tǒng)計(jì)值并展示(truncate=False表示值全部顯示，不進(jìn)行縮略)print("計(jì)算統(tǒng)計(jì)值并展示:")df.select(summarizer.summary(df.features,df.weight)).show(truncate=False)#不帶權(quán)重的統(tǒng)計(jì)print("不帶權(quán)重的統(tǒng)計(jì):")df.select(summarizer.summary(df.features)).show(truncate=False)#統(tǒng)計(jì)帶有權(quán)重的均值print("統(tǒng)計(jì)帶有權(quán)重的均值:")df.select(Summarizer.mean(df.features,df.weight)).show(truncate=False)#統(tǒng)計(jì)不帶有權(quán)重的均值print("統(tǒng)計(jì)不帶有權(quán)重的均值:")df.select(Summarizer.mean(df.features)).show(truncate=False)4基本統(tǒng)計(jì)匯總統(tǒng)計(jì)print("不帶權(quán)重的統(tǒng)計(jì):")df.select(summarizer.summary(df.features)).show(truncate=False)#統(tǒng)計(jì)帶有權(quán)重的均值print("統(tǒng)計(jì)帶有權(quán)重的均值:")df.select(Summarizer.mean(df.features,df.weight)).show(truncate=False)#統(tǒng)計(jì)不帶有權(quán)重的均值print("統(tǒng)計(jì)不帶有權(quán)重的均值:")df.select(Summarizer.mean(df.features)).show(truncate=False)4基本統(tǒng)計(jì)相關(guān)性分析相關(guān)分析是研究?jī)蓚€(gè)或兩個(gè)以上的變量間的相關(guān)關(guān)系的統(tǒng)計(jì)分析方法。例如，人的身高和體重之間、空氣中的相對(duì)濕度與降雨量之間的相關(guān)關(guān)系。在一段時(shí)期內(nèi)商品房?jī)r(jià)格隨經(jīng)濟(jì)水平上升而上升，這說(shuō)明兩指標(biāo)間是正相關(guān)關(guān)系；而在另一時(shí)期，隨著經(jīng)濟(jì)水平進(jìn)一步發(fā)展，出現(xiàn)商品房?jī)r(jià)格下降的現(xiàn)象，兩指標(biāo)間就是負(fù)相關(guān)關(guān)系。相關(guān)系數(shù)是變量間關(guān)聯(lián)程度的最基本測(cè)度之一，如果我們想知道兩個(gè)變量之間的相關(guān)性，那么我們就可以計(jì)算相關(guān)系數(shù)來(lái)進(jìn)行判定。相關(guān)系數(shù)r的取值在-1和1之間。正相關(guān)時(shí)，r值在0和1之間，散點(diǎn)圖是斜向上的，這時(shí)一個(gè)變量增加，另一個(gè)變量也增加；負(fù)相關(guān)時(shí)，r值在-1和0之間，散點(diǎn)圖是斜向下的，此時(shí)一個(gè)變量增加，另一個(gè)變量將減少。r的絕對(duì)值越接近1，兩變量的關(guān)聯(lián)程度越強(qiáng)，r的絕對(duì)值越接近0，兩變量的關(guān)聯(lián)程度越弱。目前Spark支持兩種相關(guān)性系數(shù)：皮爾遜（Pearson）相關(guān)系數(shù)和斯皮爾曼（Spearman）相關(guān)系數(shù)（Spearman）。4基本統(tǒng)計(jì)相關(guān)性分析

4基本統(tǒng)計(jì)相關(guān)性分析

4基本統(tǒng)計(jì)相關(guān)性分析2.Spearman相關(guān)系數(shù)>>>df.show()+--------------------+|features|+--------------------+|(4,[0,3],[1.0,-2.0])||[4.0,5.0,0.0,3.0]||[6.0,7.0,0.0,8.0]||(4,[0,3],[9.0,1.0])|+--------------------+#每個(gè)元素表示左右vector列表取對(duì)應(yīng)兩兩vector的相關(guān)系數(shù)>>>r1=Correlation.corr(df,"features").head()#默認(rèn)Pearson相關(guān)系數(shù)4基本統(tǒng)計(jì)相關(guān)性分析2.Spearman相關(guān)系數(shù)>>>print("Pearsoncorrelationmatrix:\n"+str(r1[0]))Pearsoncorrelationmatrix:DenseMatrix([[1.,0.05564149,nan,0.40047142],[0.05564149,1.,nan,0.91359586],[nan,nan,1.,nan],[0.40047142,0.91359586,nan,1.]])>>>r2=Correlation.corr(df,"features","spearman").head()#求Spearman相關(guān)系數(shù)>>>print("Spearmancorrelationmatrix:\n"+str(r2[0]))Spearmancorrelationmatrix:DenseMatrix([[1.,0.10540926,nan,0.4],[0.10540926,1.,nan,0.9486833],[nan,nan,1.,nan],[0.4,0.9486833,nan,1.]])4基本統(tǒng)計(jì)分層抽樣當(dāng)總體由明顯差別的幾部分組成時(shí)，常采用分層抽樣，即將總體中各個(gè)個(gè)體首先按某種特征分成若干個(gè)互不重疊的幾部分，每一部分叫做層，然后在各層中按層在總體中所占的比例進(jìn)行簡(jiǎn)單隨機(jī)抽樣或系統(tǒng)抽樣。1.sampleByKey()方法sampleByKey()方法需要作用于一個(gè)鍵值對(duì)數(shù)組，其中鍵-值對(duì)的鍵用于分類(lèi)（分層次），sampleByKey()方法通過(guò)設(shè)置抽取函數(shù)fractions來(lái)定義分類(lèi)條件和采樣幾率。//創(chuàng)建鍵-值對(duì)類(lèi)型的RDD>>>data=sc.parallelize([("female","WangLi"),("female","LiuTao"),("female","LiQian"),("female","TangLi"),("female","FeiFei"),("male","WangQiang"),("male","WangChao"),("male","LiHua"),("male","GeLin"),("male","LiJian")])#指定不同鍵的抽取比例>>>fractions={"female":0.6,"male":0.4}4基本統(tǒng)計(jì)分層抽樣1.sampleByKey()方法#用sampleByKey()方法進(jìn)行抽樣：>>>approxSample=data.sampleByKey(withReplacement=False,fractions=fractions)>>>approxSample.foreach(print)#輸出抽樣結(jié)果('female','WangLi')('female','LiuTao')('female','LiQian')('female','FeiFei')('male','WangChao')4基本統(tǒng)計(jì)生成隨機(jī)數(shù)RandomRDDs工具集提供了生成雙精度隨機(jī)數(shù)RDD和向量RDD的工廠方法，可指定生成隨機(jī)數(shù)的分布模式。下面的例子中生成一個(gè)雙精度隨機(jī)數(shù)RDD，其元素值服從標(biāo)準(zhǔn)正態(tài)分布N(0,1)，然后將其映射到N(1,4)。>>>frompyspark.mllib.randomimportRandomRDDs#生成10000個(gè)服從正態(tài)分配N(xiāo)(0,1)的RDD，分區(qū)的個(gè)數(shù)為5>>>NRDD=RandomRDDs.normalRDD(sc,10000,5)#轉(zhuǎn)化生成服從N(1,4)正態(tài)分布的RDD>>>NRDD1=NRDD.map(lambdax:1.0+2.0*x)4基本統(tǒng)計(jì)核密度估計(jì)核密度估計(jì)不利用有關(guān)數(shù)據(jù)分布的先驗(yàn)知識(shí)，對(duì)數(shù)據(jù)分布不附加任何假定，是一種從數(shù)據(jù)樣本本身出發(fā)研究數(shù)據(jù)分布特征的方法。核密度估計(jì)是在概率論中用來(lái)估計(jì)未知的密度函數(shù)，采用平滑的峰值函數(shù)(“核”)來(lái)擬合觀察到的數(shù)據(jù)點(diǎn)，從而對(duì)真實(shí)的概率分布曲線進(jìn)行模擬。>>>frompyspark.mllib.statimportKernelDensity>>>data=sc.parallelize([1.0,1.0,1.0,2.0,3.0,4.0,5.0,5.0,6.0,7.0,8.0,9.0,9.0])#用樣本數(shù)據(jù)和高斯核的標(biāo)準(zhǔn)差構(gòu)建密度估計(jì)器>>>kd=KernelDensity()>>>kd.setSample(data)#setBandwidth()設(shè)置高斯核的寬度，可以看做是高斯核的標(biāo)準(zhǔn)差>>>kd.setBandwidth(3.0)4基本統(tǒng)計(jì)核密度估計(jì)#用構(gòu)造的核密度估計(jì)器kd對(duì)給定數(shù)據(jù)進(jìn)行核估計(jì)>>>densities=kd.estimate([-1.0,2.0,5.0])>>>densitiesarray([0.04145944,0.07902017,0.0896292])輸出結(jié)果表示：在-1.0、2.0、5.0等樣本點(diǎn)上，估算的概率密度函數(shù)值分別是0.04145944、0.07902017和0.0896292。

SparkMLlib機(jī)器學(xué)習(xí)1 MLlib機(jī)器學(xué)習(xí)庫(kù)2 MLlib基本數(shù)據(jù)類(lèi)型3 機(jī)器學(xué)習(xí)流水線4 基本統(tǒng)計(jì)5 特征提取、轉(zhuǎn)換和選擇6分類(lèi)算法7回歸算法8聚類(lèi)算法9協(xié)同過(guò)濾推薦算法5特征提取、轉(zhuǎn)換和選擇特征提取

5特征提取、轉(zhuǎn)換和選擇特征提取

5特征提取、轉(zhuǎn)換和選擇特征提取#使用Tokenizer構(gòu)建分詞器>>>tokenizer=Tokenizer(inputCol="sentence",outputCol="words")#通過(guò)分詞器對(duì)句子進(jìn)行分詞>>>wordsData=tokenizer.transform(sentenceData)>>>wordsData.show()+-----+--------------------+--------------------+|label|sentence|words|+-----+--------------------+--------------------+|0.0|HiIheardabout...|[hi,i,heard,ab...||0.0|IwishJavacould...|[i,wish,java,c...||1.0|Logisticregressi...|[logistic,regres...|+-----+--------------------+--------------------+從上面的輸出結(jié)果可以看出，transform方法把每個(gè)句子拆分成一個(gè)個(gè)單詞，這些單詞構(gòu)成一個(gè)“詞袋”（里面裝了很多個(gè)單詞）。5特征提取、轉(zhuǎn)換和選擇特征提取#使用HashingTF構(gòu)建詞頻統(tǒng)計(jì)器，這里設(shè)置哈希表的桶數(shù)為20>>>hashingTF=HashingTF(inputCol="words",outputCol="rawFeatures",numFeatures=20)#通過(guò)詞頻統(tǒng)計(jì)器來(lái)統(tǒng)計(jì)詞頻，把每個(gè)“詞袋”哈希成特征向量>>>featurizedData=hashingTF.transform(wordsData)>>>featurizedData.show()+-----+--------------------+--------------------+--------------------+|label|sentence|words|rawFeatures|+-----+--------------------+--------------------+--------------------+|0.0|HiIheardabout...|[hi,i,heard,ab...|(20,[6,8,13,16],[...||0.0|IwishJavacould...|[i,wish,java,c...|(20,[0,2,7,13,15,...||1.0|Logisticregressi...|[logistic,regres...|(20,[3,4,6,11,19]...|+-----+--------------------+--------------------+--------------------+可以看出，“詞袋”中的每一個(gè)單詞被哈希成了一個(gè)不同的索引值。5特征提取、轉(zhuǎn)換和選擇特征提取#使用IDF構(gòu)建評(píng)估器>>>idf=IDF(inputCol="rawFeatures",outputCol="features")#在featurizedData上應(yīng)用fit()方法訓(xùn)練idf得到一個(gè)IDF模型idfModel>>>idfModel=idf.fit(featurizedData)#調(diào)用idfModel的transform方法，得到每一個(gè)單詞對(duì)應(yīng)的TF-IDF度量值>>>rescaledData=idfModel.transform(featurizedData)#查看TF-IDF值>>>rescaledData.select("label","features").show()+-----+--------------------+|label|features|+-----+--------------------+|0.0|(20,[6,8,13,16],[...||0.0|(20,[0,2,7,13,15,...||1.0|(20,[3,4,6,11,19]...|+-----+--------------------+5特征提取、轉(zhuǎn)換和選擇特征提取>>>rescaledData.show()+-----+--------------------+--------------------+--------------------+-----------------+|label|sentence|words|rawFeatures|features|+-----+--------------------+--------------------+--------------------+-----------------+|0.0|HiIheardabout...|[hi,i,heard,ab...|(20,[6,8,13,16],[...|(20,[6,8,13,16],[||0.0|IwishJavacould...|[i,wish,java,c...|(20,[0,2,7,13,15,...|(20,[0,2,7,13,15,||1.0|Logisticregressi.|[logistic,regres..|(20,[3,4,6,11,19]...|(20,[3,4,6,11,19]..|+-----+--------------------+--------------------+--------------------+-----------------+5特征提取、轉(zhuǎn)換和選擇特征轉(zhuǎn)換特征轉(zhuǎn)換只是將特征從一種形式轉(zhuǎn)換為另一種表示形式。機(jī)器學(xué)習(xí)中經(jīng)常需要對(duì)特征進(jìn)行轉(zhuǎn)換的原因：數(shù)據(jù)類(lèi)型不適合要用到的機(jī)器學(xué)習(xí)算法，通過(guò)轉(zhuǎn)換而得到新特征可以消除原特征之間的相關(guān)性或減少冗余。1.Binarizer（二值化）Binarizer（二值化）可以根據(jù)給定的閾值，將數(shù)值型特征的值轉(zhuǎn)化為0-1的二元特征值，大于閾值的特征值被二元化為1.0，小于等于閾值的特征值被二元化為0.0。frompyspark.ml.featureimportBinarizerfrompyspark.sqlimportSparkSessionspark=SparkSession.builder.getOrCreate()5特征提取、轉(zhuǎn)換和選擇特征轉(zhuǎn)換1.Binarizer（二值化）continuousDataFrame=spark.createDataFrame([(0,0.1),(1,0.8),(2,0.2)],["id","feature"])binarizer=Binarizer(threshold=0.5,inputCol="feature",outputCol="binarized_feature")binarizedDataFrame=binarizer.transform(continuousDataFrame)#二值化轉(zhuǎn)換print("BinarizeroutputwithThreshold=%f"%binarizer.getThreshold())binarizedDataFrame.show()5特征提取、轉(zhuǎn)換和選擇特征轉(zhuǎn)換1.Binarizer（二值化）continuousDataFrame=spark.createDataFrame([(0,0.1),(1,0.8),(2,0.2)],["id","feature"])binarizer=Binarizer(threshold=0.5,inputCol="feature",outputCol="binarized_feature")binarizedDataFrame=binarizer.transform(continuousDataFrame)#二值化轉(zhuǎn)換print("BinarizeroutputwithThreshold=%f"%binarizer.getThreshold())binarizedDataFrame.show()5特征提取、轉(zhuǎn)換和選擇特征選擇特征選擇指的是從特征向量中選擇出好的特征，好的特征選擇能夠提升機(jī)器學(xué)習(xí)模型的性能。特征選擇的主要功能：減少特征向量的特征數(shù)量，使模型泛化能力更強(qiáng)，減少過(guò)擬合。下面介紹pyspark.ml模塊的幾種特征選擇方法。1.VectorSlicerVectorSlicer是一種轉(zhuǎn)換器，它獲取一個(gè)特征向量并輸出一個(gè)由原始特征的子集組成新的特征向量。它對(duì)從列向量中提取特征非常有用。VectorSlicer接受帶有指定索引的向量列，然后通過(guò)這些索引進(jìn)行篩選得到一個(gè)新的向量列。有兩種類(lèi)型的索引：整數(shù)索引和字符串索引。選擇特征的幾個(gè)規(guī)則：去掉取值變化小的特征；單變量特征選擇，對(duì)每一個(gè)特征進(jìn)行測(cè)試，衡量該特征和響應(yīng)變量之間的關(guān)系，根據(jù)得分扔掉不好的特征，對(duì)于回歸和分類(lèi)問(wèn)題可以采用卡方檢驗(yàn)等方式對(duì)特征進(jìn)行測(cè)試；用相關(guān)系數(shù)法選取相關(guān)性大的特征。frompyspark.ml.featureimportVectorSlicerfrompyspark.ml.linalgimportVectorsfrompyspark.sql.typesimportRow5特征提取、轉(zhuǎn)換和選擇特征選擇1.VectorSlicerfrompyspark.sqlimportSparkSessionspark=SparkSession.builder.getOrCreate()df=spark.createDataFrame([Row(userFeatures=Vectors.sparse(5,{0:-2.0,1:2.3,2:3.3})),Row(userFeatures=Vectors.dense([-2.0,2.3,0.0,5.3,0.7]))])#indices=[1,3]表示對(duì)向量列選擇第索引為1和3的特征slicer=VectorSlicer(inputCol="userFeatures",outputCol="features",indices=[1,3])output=slicer.transform(df)#轉(zhuǎn)換數(shù)據(jù)output.select("userFeatures","features").show(truncate=False)運(yùn)行上述程序代碼，得到的輸出結(jié)果如下。

SparkMLlib機(jī)器學(xué)習(xí)1 MLlib機(jī)器學(xué)習(xí)庫(kù)2 MLlib基本數(shù)據(jù)類(lèi)型3 機(jī)器學(xué)習(xí)流水線4 基本統(tǒng)計(jì)5 特征提取、轉(zhuǎn)換和選擇6分類(lèi)算法7回歸算法8聚類(lèi)算法9協(xié)同過(guò)濾推薦算法6分類(lèi)算法邏輯回歸二分類(lèi)邏輯回歸分析（LogisticRegression）是用于處理因變量為分類(lèi)變量的回歸分析。邏輯回歸分析根據(jù)因變量取值類(lèi)別不同，又可以分為二分類(lèi)回歸分析和多分類(lèi)回歸分析。二分類(lèi)回歸模型中，因變量Y只有“是、否”兩個(gè)取值，記為1和0，而多分類(lèi)回歸模型中因變量可以取多個(gè)值。

考慮二分類(lèi)問(wèn)題，其輸出標(biāo)記

，而線性回歸模型產(chǎn)生的預(yù)測(cè)值

是連續(xù)的實(shí)數(shù)值，于是，我們需要將實(shí)數(shù)值

轉(zhuǎn)換為0或1，即需要選擇一個(gè)函數(shù)將

映射到0或1，這樣的函數(shù)常選用對(duì)數(shù)幾率函數(shù)，也稱(chēng)為Sigmoid函數(shù)，其函數(shù)表達(dá)式為：6分類(lèi)算法決策樹(shù)分類(lèi)決策樹(shù)簡(jiǎn)單來(lái)說(shuō)就是帶有判決規(guī)則的一種樹(shù)，可以依據(jù)樹(shù)中的判決規(guī)則來(lái)預(yù)測(cè)未知樣本的類(lèi)別和值。決策樹(shù)就是通過(guò)樹(shù)結(jié)構(gòu)來(lái)表示各種可能的決策路徑，以及每個(gè)路徑的結(jié)果。一棵決策樹(shù)一般包含一個(gè)根結(jié)點(diǎn)、若干個(gè)內(nèi)部結(jié)點(diǎn)和若干個(gè)葉子結(jié)點(diǎn)：（1）葉子結(jié)點(diǎn)對(duì)應(yīng)于決策結(jié)果；（2）每個(gè)內(nèi)部結(jié)點(diǎn)對(duì)應(yīng)于一個(gè)屬性測(cè)試，每個(gè)內(nèi)部結(jié)點(diǎn)包含的樣本集合根據(jù)屬性測(cè)試的結(jié)果被劃分到它的子結(jié)點(diǎn)中；（3）根結(jié)點(diǎn)包含全部訓(xùn)練樣本；（4）從根結(jié)點(diǎn)到每個(gè)葉子結(jié)點(diǎn)的路徑對(duì)應(yīng)了一條決策規(guī)則。下面借助sample_libsvm_data.txt數(shù)據(jù)集來(lái)演示DecisionTreeClassifier實(shí)現(xiàn)決策樹(shù)分類(lèi)。frompyspark.sqlimportSparkSessionfrompyspark.mlimportPipelinefrompyspark.ml.classificationimportDecisionTreeClassifier6分類(lèi)算法決策樹(shù)分類(lèi)frompyspark.ml.evaluationimportMulticlassClassificationEvaluatorfrompyspark.ml.featureimportStringIndexer,VectorIndexerspark=SparkSession.builder.getOrCreate()#決策樹(shù)分類(lèi)器defdecisionTreeClassifier(data):#加載LIBSVM格式的數(shù)據(jù)集

data=spark.read.format("libsvm").load(data)#將字符串標(biāo)簽（不是則先轉(zhuǎn)為字符串）編碼為整數(shù)索引（頻率從高到低的標(biāo)簽轉(zhuǎn)為0,1,...）#逆操作：IndexToStringlabelIndexer=StringIndexer(inputCol="label",outputCol="indexedLabel").fit(data)#根據(jù)maxCategories自動(dòng)識(shí)別分類(lèi)特征（如果某個(gè)特征不重復(fù)的值小于等于maxCategories#則該特征轉(zhuǎn)為分類(lèi)特征）

featureIndexer=VectorIndexer(inputCol="features",outputCol="indexedFeatures",maxCategories=4).fit(data)6分類(lèi)算法決策樹(shù)分類(lèi)#將數(shù)據(jù)集劃分為訓(xùn)練集和測(cè)試集，測(cè)試集占30%(trainingData,testData)=data.randomSplit([0.7,0.3])#創(chuàng)建決策樹(shù)模型

dt=DecisionTreeClassifier(labelCol="indexedLabel",featuresCol="indexedFeatures")pipeline=Pipeline(stages=[labelIndexer,featureIndexer,dt])model=pipeline.fit(trainingData)#訓(xùn)練模型

predictions=model.transform(testData)#進(jìn)行預(yù)測(cè)

print("展示預(yù)測(cè)結(jié)果的前5行數(shù)據(jù):")predictions.select("prediction","indexedLabel","features").show(5)#選擇預(yù)測(cè)標(biāo)簽、真實(shí)標(biāo)簽，計(jì)算測(cè)試誤差

evaluator=MulticlassClassificationEvaluator(labelCol="indexedLabel",predictionCol="prediction",metricName="accuracy")accuracy=evaluator.evaluate(predictions)print("測(cè)試誤差=%g"%(1.0-accuracy))treeModel=model.stages[2]print('treeModel摘要:',treeModel)#模型摘要6分類(lèi)算法決策樹(shù)分類(lèi)

if__name__=='__main__':decisionTreeClassifier(data="file:///usr/local/spark/data/mllib/sample_libsvm_data.txt")運(yùn)行上述程序代碼，得到的輸出結(jié)果如下。展示預(yù)測(cè)結(jié)果的前5行數(shù)據(jù):+----------+------------+--------------------+|prediction|indexedLabel|features|+----------+------------+--------------------+|1.0|1.0|(692,[121,122,123...||1.0|1.0|(692,[122,123,148...||1.0|1.0|(692,[123,124,125...||1.0|1.0|(692,[124,125,126...||1.0|1.0|(692,[124,125,126...|+----------+------------+--------------------+onlyshowingtop5rows

SparkMLlib機(jī)器學(xué)習(xí)1 MLlib機(jī)器學(xué)習(xí)庫(kù)2 MLlib基本數(shù)據(jù)類(lèi)型3 機(jī)器學(xué)習(xí)流水線4 基本統(tǒng)計(jì)5 特征提取、轉(zhuǎn)換和選擇6分類(lèi)算法7回歸算法8聚類(lèi)算法9協(xié)同過(guò)濾推薦算法7回歸算法回歸決策樹(shù)回歸決策樹(shù)樹(shù)是用于回歸的決策樹(shù)模型，回歸決策樹(shù)主要指CART決策樹(shù)。CART全稱(chēng)為ClassificationandRegressionTree，即分類(lèi)與回歸樹(shù)，CART既可以用于分類(lèi)還可以用于回歸。CART分類(lèi)樹(shù)算法使用基尼系數(shù)來(lái)代替信息增益率，基尼系數(shù)代表了樣本數(shù)據(jù)集的不純度，基尼系數(shù)越小，不純度越低，特征越好，這和信息增益（率）相反。CART回歸樹(shù)算法，使用的是平方誤差最小準(zhǔn)則。>>>frompyspark.ml.regressionimportDecisionTreeRegressor#創(chuàng)建DecisionTreeRegressor模型，設(shè)置特征例和預(yù)測(cè)列>>>dt=DecisionTreeRegressor(featuresCol='features',labelCol='PE')>>>dt_model=dt.fit(train_df)#用train_df訓(xùn)練模型，實(shí)際上得到一個(gè)轉(zhuǎn)換器#調(diào)用transform()預(yù)測(cè)，得到一個(gè)DataFrame>>>dt_predictions=dt_model.transform(test_df)

7回歸算法回歸決策樹(shù)>>>dt_predictions.show(5)+--------------------+------+-----------------+|features|PE|prediction|+--------------------+------+-----------------+|[1.81,39.42,1026....|490.55|486.7955675675676||[2.8,39.64,1011.0...|482.66|486.7955675675676||[3.21,38.44,1016....|491.35|486.7955675675676||[3.91,35.47,1016....|488.67|486.7955675675676||[3.95,35.47,1017....|488.64|486.7955675675676|+--------------------+------+-----------------+7回歸算法回歸決策樹(shù)#模型評(píng)估>>>frompyspark.ml.evaluationimportRegressionEvaluator>>>dt_evaluator=RegressionEvaluator(labelCol='PE',predictionCol="prediction",metricName="rmse")>>>rmse=dt_evaluator.evaluate(dt_predictions)>>>print("RootMeanSquaredError(RMSE)ontestdata=%g"%rmse)RootMeanSquaredError(RMSE)ontestdata=4.38643#各特征重要程度>>>dt_model.featureImportancesSparseVector(4,{0:0.9562,1:0.04,2:0.0017,3:0.0021})

SparkMLlib機(jī)器學(xué)習(xí)1 MLlib機(jī)器學(xué)習(xí)庫(kù)2 MLlib基本數(shù)據(jù)類(lèi)型3 機(jī)器學(xué)習(xí)流水線4 基本統(tǒng)計(jì)5 特征提取、轉(zhuǎn)換和選擇6分類(lèi)算法7回歸算法8聚類(lèi)算法9協(xié)同過(guò)濾推薦算法8聚類(lèi)算法K均值聚類(lèi)算法K均值（K-means）聚類(lèi)算法也被稱(chēng)為k-平均聚類(lèi)算法，是一種最廣泛使用的聚類(lèi)算法。K均值用質(zhì)心來(lái)表示一個(gè)簇，質(zhì)心就是一組數(shù)據(jù)對(duì)象點(diǎn)的平均值。K均值算法以k為輸入?yún)?shù)，將n個(gè)數(shù)據(jù)對(duì)象劃分為k個(gè)簇，使得簇內(nèi)數(shù)據(jù)對(duì)象具有較高的相似度。K均值聚類(lèi)的算法思想：從包含n個(gè)數(shù)據(jù)對(duì)象的數(shù)據(jù)集中隨機(jī)的選擇k個(gè)對(duì)象，每個(gè)對(duì)象代表一個(gè)簇的平均值或質(zhì)心或中心，其中k是用戶(hù)指定的參數(shù)，即所期望的要?jiǎng)澐殖傻拇氐膫€(gè)數(shù)；對(duì)剩余的每個(gè)數(shù)據(jù)對(duì)象點(diǎn)根據(jù)其與各個(gè)簇中心的距離，將它指派到最近的簇；然后，根據(jù)指派到簇的數(shù)據(jù)對(duì)象點(diǎn)，更新每個(gè)簇的中心；重復(fù)指派和更新步驟，直到簇不發(fā)生變化，或直到中心不發(fā)生變化，或度量聚類(lèi)質(zhì)量的目標(biāo)函數(shù)收斂。8聚類(lèi)算法K均值聚類(lèi)算法下面給出KMeans的K均值聚類(lèi)實(shí)現(xiàn)代碼。frompyspark.sqlimportSparkSessionfrompyspark.ml.featureimportVectorAssemblerfrompyspark.ml.clusteringimportKMeansspark=SparkSession.builder.getOrCreate()df=spark.read.csv("file:///home/hadoop/iris.csv",header=True,inferSchema=True)vec_assembler=VectorAssembler(inputCols=['Sep_len','Sep_wid','Pet_len','Pet_wid'],outputCol='features')#轉(zhuǎn)換數(shù)據(jù)后，在返回的DataFrame中添加了features列features_df=vec_assembler.transform(df).select('Iris_type','features')#創(chuàng)建模型，也可表示成kmeans=KMeans(k=3,seed=1,predictionCol="prediction"kmeans=KMeans().setK(3).setSeed(1).setPredictionCol('prediction')model=kmeans.fit(features_df)8聚類(lèi)算法K均值聚類(lèi)算法#進(jìn)行預(yù)測(cè)，生成帶有預(yù)測(cè)簇標(biāo)簽的數(shù)據(jù)集predictions=model.transform(features_df)predictions.show(55)運(yùn)行上述程序代碼，得到的部分輸出結(jié)果如。+---------------+-----------------+----------+|Iris_type|features|prediction|+---------------+-----------------+----------+|Iris-setosa|[5.1,3.5,1.4,0.2]|1||Iris-setosa|[4.9,3.0,1.4,0.2]|1|......|Iris-setosa|[5.3,3.7,1.5,0.2]|1||Iris-setosa|[5.0,3.3,1.4,0.2]|1||Iris-versicolor|[7.0,3.2,4.7,1.4]|0|+---------------+-----------------+----------+

SparkMLlib機(jī)器學(xué)習(xí)1 MLlib機(jī)器學(xué)習(xí)庫(kù)2 MLlib基本數(shù)據(jù)類(lèi)型3 機(jī)器學(xué)習(xí)流水線4 基本統(tǒng)計(jì)5 特征提取、轉(zhuǎn)換和選擇6分類(lèi)算法7回歸算法8聚類(lèi)算法9協(xié)同過(guò)濾推薦算法9協(xié)同過(guò)濾推薦算法協(xié)同過(guò)濾推薦指的是利用某興趣相投、擁有共同經(jīng)驗(yàn)的群體的喜好來(lái)推薦用戶(hù)感興趣的信息。協(xié)同過(guò)濾推薦算法主要分為基于用戶(hù)的協(xié)同過(guò)濾推薦和基于物品的協(xié)同過(guò)濾推薦。1.基于用戶(hù)的協(xié)同過(guò)濾推薦基于用戶(hù)的協(xié)同過(guò)濾推薦，通過(guò)不同用戶(hù)對(duì)物品的相似評(píng)分來(lái)發(fā)現(xiàn)在某一方面相似的用戶(hù)組，然后根據(jù)相似的用戶(hù)組的喜好產(chǎn)生向目標(biāo)用戶(hù)推薦的內(nèi)容。基本原理就是利用用戶(hù)訪問(wèn)行為的相似性來(lái)互相推薦用戶(hù)可能感興趣的內(nèi)容，如圖所示9協(xié)同過(guò)濾推薦算法協(xié)同過(guò)濾推薦指的是利用某興趣相投、擁有共同經(jīng)驗(yàn)的群體的喜好來(lái)推薦用戶(hù)感興趣的信息。協(xié)同過(guò)濾推薦算法主要分為基于用戶(hù)的協(xié)同過(guò)濾推薦和基于物品的協(xié)同過(guò)濾推薦。1.基于用戶(hù)的協(xié)同過(guò)濾推薦基于用戶(hù)的協(xié)同過(guò)濾推薦，通過(guò)不同用戶(hù)對(duì)物品的相似評(píng)分來(lái)發(fā)現(xiàn)在某一方面相似的用戶(hù)組，然后根據(jù)相似的用戶(hù)組的喜好產(chǎn)生向目標(biāo)用戶(hù)推薦的內(nèi)容。基本原理就是利用用戶(hù)訪問(wèn)行為的相似性來(lái)互相推薦用戶(hù)可能感興趣的內(nèi)容，如圖所示右圖示意出基于用戶(hù)的協(xié)同過(guò)濾推薦機(jī)制的基本原理：假設(shè)用戶(hù)A喜歡物品A、物品C，用戶(hù)B喜歡物品B，用戶(hù)C喜歡物品A、物品C和物品D；從這些用戶(hù)的歷史喜好信息中，可以發(fā)現(xiàn)用戶(hù)A和用戶(hù)C的偏好是比較類(lèi)似的，同時(shí)用戶(hù)C還喜歡物品D，那么可以推斷用戶(hù)A可能也喜歡物品D，因此可以將物品D推薦給用戶(hù)A。9協(xié)同過(guò)濾推薦算法根據(jù)用戶(hù)對(duì)物品的評(píng)分來(lái)評(píng)測(cè)物品之間的相似性，然后根據(jù)物品的相似性向目標(biāo)用戶(hù)推薦感興趣的物品，如圖所示。右圖表明基于物品的協(xié)同過(guò)濾推薦的基本原理：用戶(hù)A、用戶(hù)B、用戶(hù)C都喜歡物品A，用戶(hù)A、用戶(hù)B都喜歡物品C，物品B只有用戶(hù)B都喜歡，由此得出物品A與物品C比較類(lèi)似，喜歡物品A的都喜歡物品C，基于這個(gè)判斷用戶(hù)C可能也喜歡物品C，所以將物品C推薦給用戶(hù)C。9協(xié)同過(guò)濾推薦算法ALS是AlternatingLeastSquares的縮寫(xiě)，意為交替最小二乘法，該方法常用于基于矩陣分解的推薦系統(tǒng)中。如將用戶(hù)(user)對(duì)商品(item)的評(píng)分矩陣分解為兩個(gè)矩陣：一個(gè)是用戶(hù)特征矩陣U，一個(gè)是商品特征矩陣V。在MLlib的ALS算法中，首先隨機(jī)生成U矩陣或者V矩陣，之后固定一個(gè)矩陣，去求取另一個(gè)未隨機(jī)化的矩陣如圖7-6所示。簡(jiǎn)單地說(shuō)，先固定U矩陣，求取V，然后再固定V矩陣再求取U矩陣，一直這樣交替迭代計(jì)算直到誤差達(dá)到一定的閾值條件或者達(dá)到迭代次數(shù)的上限。ALS交替最小二乘協(xié)同過(guò)濾推薦9協(xié)同過(guò)濾推薦算法下面給出ALS的用法舉例，首先加載Spark自帶的評(píng)分?jǐn)?shù)據(jù)sample_movielens_ratings.txt(位于/usr/local/spark/data/mllib/als目錄下)，其中每行數(shù)據(jù)由一個(gè)用戶(hù)ID、一個(gè)產(chǎn)品ID和一個(gè)評(píng)分組成，文件中部分?jǐn)?shù)據(jù)如下所示。0::2::3::3::1::5::2::9::4::11::1::1424380312具體實(shí)現(xiàn)代碼如下。frompyspark.sqlimportSparkSessionfrompyspark.ml.evaluationimportRegressionEvaluatorfrompyspark.ml.recommendationimportALSfrompyspark.sqlimportRowspark=SparkSession.builder.getOrCreate()ALS交替最小二乘協(xié)同過(guò)濾推薦9協(xié)同過(guò)濾推薦算法lines=spark.read.text("file:///usr/local/spark/data/mllib/als/sample_movielens_ratings.txt").rddparts=lines.map(lambdarow:row.value.split("::"))ratingsRDD=parts.map(lambdap:Row(userId=int(p[0]),movieId=int(p[1]),rating=float(p[2]),timestamp=int(p[3])))ratings=spark.createDataFrame(ratingsRDD)#創(chuàng)建DataFrame對(duì)象(training,test)=ratings.randomSplit([0.8,0.2])#分割為訓(xùn)練集和測(cè)試集#創(chuàng)建ALS推薦模型als=ALS(maxIter=5,regParam=0.01,userCol="userId",itemCol="movieId",ratingCol="rating",coldStartStrategy="drop")model=als.fit(training)#訓(xùn)練模型predictions=model.transform(test)#預(yù)測(cè)ALS交替最小二乘協(xié)同過(guò)濾推薦9協(xié)同過(guò)濾推薦算法#計(jì)算模型在test數(shù)據(jù)集上的均方根誤差RMSE來(lái)評(píng)估模型evaluator=RegressionEvaluator(metricName="rmse",labelCol="rating",predictionCol="prediction")rmse=evaluator.evaluate(predictions)print("Root-mean-squareerror="+str(rmse))print('為每個(gè)用戶(hù)推薦10部電影:')userRecs=model.recommendForAllUsers(10)userRecs.show()print('為每部電影推薦10個(gè)用戶(hù):')movieRecs=model.recommendForAllItems(10)movieRecs.show()print('為具體用戶(hù)推薦10部電影:')users=ratings.se