SparkMLlib第9章目

錄01基于大數(shù)據(jù)的機(jī)器學(xué)習(xí)02機(jī)器學(xué)習(xí)庫Mllib概述03基本數(shù)據(jù)類型04基本統(tǒng)計工具05機(jī)器學(xué)習(xí)流水線06特征抽取、轉(zhuǎn)換和選擇目

錄07分類算法08聚類算法09頻繁模式挖掘算法10協(xié)同過濾算法11模型選擇和超參數(shù)調(diào)整基于大數(shù)據(jù)的機(jī)器學(xué)習(xí)9.1基于大數(shù)據(jù)的機(jī)器學(xué)習(xí)機(jī)器學(xué)習(xí)機(jī)器學(xué)習(xí)可以看做是一門人工智能的科學(xué)，該領(lǐng)域的主要研究對象是人工智能。機(jī)器學(xué)習(xí)利用數(shù)據(jù)或以往的經(jīng)驗，以此優(yōu)化計算機(jī)程序的性能標(biāo)準(zhǔn)機(jī)器學(xué)習(xí)強(qiáng)調(diào)三個關(guān)鍵詞：算法、經(jīng)驗、性能9.1基于大數(shù)據(jù)的機(jī)器學(xué)習(xí)傳統(tǒng)機(jī)器學(xué)習(xí)算法輸入特征提取特征傳統(tǒng)機(jī)器學(xué)習(xí)算法輸出傳統(tǒng)機(jī)器學(xué)習(xí)有限的數(shù)據(jù)量進(jìn)行計算傳統(tǒng)機(jī)器學(xué)習(xí)算法，由于技術(shù)和單機(jī)存儲的限制，只能在少量數(shù)據(jù)上使用，依賴于數(shù)據(jù)抽樣9.1基于大數(shù)據(jù)的機(jī)器學(xué)習(xí)Spark機(jī)器學(xué)習(xí)大量數(shù)據(jù)機(jī)器學(xué)習(xí)訓(xùn)練大數(shù)據(jù)技術(shù)的出現(xiàn)可以支持在全量數(shù)據(jù)上進(jìn)行機(jī)器學(xué)習(xí)數(shù)

據(jù)數(shù)

據(jù)數(shù)

據(jù)......9.1基于大數(shù)據(jù)的機(jī)器學(xué)習(xí)反復(fù)讀寫磁盤的開銷磁盤IO開銷比較大的缺陷使用MapReduce對機(jī)器學(xué)習(xí)的算法進(jìn)行編寫關(guān)鍵詞迭代過程適合大量迭代計算機(jī)器學(xué)習(xí)算法涉及大量迭代計算9.1基于大數(shù)據(jù)的機(jī)器學(xué)習(xí)海量數(shù)據(jù)機(jī)器學(xué)習(xí)庫分布式的實現(xiàn)開發(fā)者算法的原理輸入?yún)?shù)含義調(diào)用API自動完成數(shù)據(jù)處理返回來處理結(jié)果9.1基于大數(shù)據(jù)的機(jī)器學(xué)習(xí)算法工程師邊寫代碼邊運行邊看結(jié)果Spark-ShellSpark-Shell的即席查詢也是一個關(guān)鍵機(jī)器學(xué)習(xí)庫Mllib概述9.2機(jī)器學(xué)習(xí)庫MLlib概述算法MLlib-機(jī)器學(xué)習(xí)庫工具M(jìn)Llib是Spark的機(jī)器學(xué)習(xí)（MachineLearning）庫，旨在簡化機(jī)器學(xué)習(xí)的工程實踐工作分類回歸聚類協(xié)同過濾包括底層的優(yōu)化原語和高層的流水線（Pipeline）API9.2機(jī)器學(xué)習(xí)庫MLlib概述協(xié)同過濾MLlib分類回歸聚類降維9.2機(jī)器學(xué)習(xí)庫MLlib概述分類回歸聚類協(xié)同過濾算法工具9.2機(jī)器學(xué)習(xí)庫MLlib概述特征提取轉(zhuǎn)化降維選擇工具特征化工具9.2機(jī)器學(xué)習(xí)庫MLlib概述流水線調(diào)整機(jī)器學(xué)習(xí)工作流構(gòu)建評估9.2機(jī)器學(xué)習(xí)庫MLlib概述持久性保存算法加載算法模型管道9.2機(jī)器學(xué)習(xí)庫MLlib概述數(shù)據(jù)處理實用工具線性代數(shù)統(tǒng)計9.2機(jī)器學(xué)習(xí)庫MLlib概述MLlib協(xié)同過濾基本統(tǒng)計回歸聚類降維分類特征抽取轉(zhuǎn)換9.2機(jī)器學(xué)習(xí)庫MLlib概述9.2機(jī)器學(xué)習(xí)庫MLlib概述機(jī)器學(xué)習(xí)庫不同的包spark.ml基于DataFrame的數(shù)據(jù)抽象spark.mllib基于RDD的數(shù)據(jù)抽象包含基于RDD的原始算法API。SparkMLlib歷史比較長，在1.0以前的版本即已經(jīng)包含了，提供的算法實現(xiàn)都是基于原始的RDD則提供了基于DataFrames高層次的API，可以用來構(gòu)建機(jī)器學(xué)習(xí)工作流（PipeLine）。MLPipeline彌補(bǔ)了原始MLlib庫的不足，向用戶提供了一個基于DataFrame的機(jī)器學(xué)習(xí)工作流式API套件

基本數(shù)據(jù)類型9.3基本數(shù)據(jù)類型01本地向量02標(biāo)注點03本地矩陣04數(shù)據(jù)源9.3.1本地向量vs稠密向量（DenseVector）稀疏向量（SparseVector）使用雙精度浮點型數(shù)組來表示每一維的元素稀疏向量則是基于一個整型索引數(shù)組和一個雙精度浮點型的值數(shù)組向量(1.0,0.0,3.0)稠密向量表示形式是[1.0,0.0,3.0]|稀疏向量形式則是(3,[0,2],[1.0,3.0])9.3.1本地向量所有本地向量都以org.apache.spark.ml.linalg.Vector為基類，DenseVector和SparseVector分別是它的兩個繼承類，故推薦使用Vectors工具類下定義的工廠方法來創(chuàng)建本地向量01需要注意的是，Scala會默認(rèn)引入scala.collection.immutable.Vector。如果要使用spark.ml包提供的向量類型，則要顯式地引入org.apache.spark.ml.linalg.Vector這個類029.3.1本地向量scala>importorg.apache.spark.ml.linalg.{Vector,Vectors}scala>valdv:Vector=Vectors.dense(2.0,0.0,8.0)dv:org.apache.spark.ml.linalg.Vector=[2.0,0.0,8.0]scala>valsv1:Vector=Vectors.sparse(3,Array(0,2),Array(2.0,8.0))sv1:org.apache.spark.ml.linalg.Vector=(3,[0,2],[2.0,8.0])scala>valsv2:Vector=Vectors.sparse(3,Seq((0,2.0),(2,8.0)))sv2:org.apache.spark.ml.linalg.Vector=(3,[0,2],[2.0,8.0])

本地向量，具體實例代碼如下9.3.2標(biāo)注點01是一種帶有標(biāo)簽的本地向量，通常用在監(jiān)督學(xué)習(xí)算法中，它可以是稠密或者是稀疏的標(biāo)注點（LabeledPoint）由于標(biāo)簽是用雙精度浮點型來存儲的，因此，標(biāo)注點類型在回歸（Regression）和分類（Classification）問題上均可使用02標(biāo)注點類型03對于二分類問題，則正樣本的標(biāo)簽為1，負(fù)樣本的標(biāo)簽為0；對于多類別的分類問題來說，標(biāo)簽則應(yīng)是一個以0開始的索引序列：0,1,2…舉例9.3.2標(biāo)注點scala>importorg.apache.spark.ml.linalg.{Vector,Vectors}scala>valdv:Vector=Vectors.dense(2.0,0.0,8.0)dv:org.apache.spark.ml.linalg.Vector=[2.0,0.0,8.0]scala>valsv1:Vector=Vectors.sparse(3,Array(0,2),Array(2.0,8.0))sv1:org.apache.spark.ml.linalg.Vector=(3,[0,2],[2.0,8.0])scala>valsv2:Vector=Vectors.sparse(3,Seq((0,2.0),(2,8.0)))sv2:org.apache.spark.ml.linalg.Vector=(3,[0,2],[2.0,8.0])

標(biāo)注點的實現(xiàn)類是org.apache.spark.ml.feature.LabeledPoint，實例如下9.3.2標(biāo)注點MLlib提供了讀取LIBSVM格式數(shù)據(jù)的支持，該格式被廣泛用于LIBSVM、LIBLINEAR等機(jī)器學(xué)習(xí)庫。在該格式下每一個帶標(biāo)簽的樣本點由以下格式表示：labelindex1:value1index2:value2index3:value3…其中，label是該樣本點的標(biāo)簽值，一系列index:value則代表了該樣本向量中所有非零元素的索引和元素值。需要特別注意的是，index是以1開始并遞增的。9.3.2標(biāo)注點scala>valexamples=spark.read.format("libsvm").|load("file:///usr/local/spark/data/MLlib/sample_libsvm_data.txt")examples:org.apache.spark.sql.DataFrame=[label:double,features:vector]

下面讀取一個LIBSVM格式文件生成向量，實例如下9.3.2標(biāo)注點scala>examples.collect().headres7:org.apache.spark.MLlib.regression.LabeledPoint=(0.0,(692,[127,128,129,130,131,154,155,156,157,158,159,181,182,183,184,185,186,187,188,189,207,208,209,210,211,212,213,214,215,216,217,235,236,237,238,239,240,241,242,243,244,245,262,263,264,265,266,267,268,269,270,271,272,273,289,290,291,292,293,294,295,296,297,300,301,302,316,317,318,319,320,321,328,329,330,343,344,345,346,347,348,349,356,357,358,371,372,373,374,384,385,386,399,400,401,412,413,414,426...],[51.0,159.0,253.0,159.0,50...

下面繼續(xù)查看加載進(jìn)來的標(biāo)注點的值，實例如下9.3.3本地矩陣稠密矩陣將所有元素的值存儲在一個列優(yōu)先（Column-major）的雙精度型數(shù)組中稀疏矩陣將非零元素以列優(yōu)先的CSC（CompressedSparseColumn）模式進(jìn)行存儲本地矩陣具有整型的行、列索引值和雙精度浮點型的元素值，它存儲在單機(jī)上MLlib支持稠密矩陣DenseMatrix和稀疏矩陣SparseMatrix兩種本地矩陣本地矩陣基類是org.apache.spark.ml.linalg.Matrix，繼承類是DenseMatrix和SparseMatrix9.3.3本地矩陣scala>importorg.apache.spark.ml.linalg.{Matrix,Matrices}importorg.apache.spark.ml.linalg.{Matrix,Matrices}scala>valdm:Matrix=Matrices.dense(3,2,Array(1.0,3.0,5.0,2.0,4.0,6.0))dm:org.apache.spark.ml.linalg.Matrix=1.02.03.04.05.06.0scala>valsm:Matrix=Matrices.sparse(3,2,Array(0,1,3),Array(0,2,1),Array(9,6,8))sm:org.apache.spark.ml.linalg.Matrix=3x2CSCMatrix(0,0)9.0(2,1)6.0(1,1)8.0

下面創(chuàng)建一個稠密矩陣，實例如下9.3.4數(shù)據(jù)源用于從一個目錄中加載圖像文件，它可以通過Java庫中的ImageIO將壓縮圖像（jpeg、png等）加載為原始圖像表示。加載的DataFrame有一個StructType列"image"，其包含以圖像模式存儲的圖像數(shù)據(jù)圖像數(shù)據(jù)源（ImageDataSource）MLlib庫提供LibSVMDatasource類來將libsvm格式的數(shù)據(jù)加載為dataframe。Libsvm是由國立臺灣大學(xué)以C++語言開發(fā)的一個開源機(jī)器學(xué)習(xí)庫，主要提供有關(guān)支持向量機(jī)的算法LIBSVM數(shù)據(jù)源9.3.4數(shù)據(jù)源StringType(表示圖像的文件路徑)IntegerType(圖像的高度)IntegerType(圖像的寬度)IntegerType(圖像的通道)IntegerType(同OpenCV兼容的圖像模式)BinaryType(按OpenCV兼容的順序排列圖像字節(jié)）modedatanChannelsheightwidthorigin9.3.4數(shù)據(jù)源scala>valdf=spark.read.format(|"image").option("dropInvalid",true).load(|"file:///usr/local/spark/data/mllib/images/origin/kittens")

第1步：讀取Spark自帶的圖像數(shù)據(jù)源格式的數(shù)據(jù)，實例如下scala>df.select("image.origin","image.width",image.height").show(truncate=false)+-----------------------------------------------------------------------+-----+------+|origin|width|height|+-----------------------------------------------------------------------+-----+------+|file:///spark/data/mllib/images/origin/kittens/54893.jpg|300|311||file:///spark/data/mllib/images/origin/kittens/DP802813.jpg|199|313||file:///spark/data/mllib/images/origin/kittens/DP153539.jpg|300|296|+-----------------------------------------------------------------------+-----+------+

第2步：輸出"image"列的origin，width和height屬性值，實例如下9.3.4數(shù)據(jù)源MLlib庫提供LibSVMDatasource類來將libsvm格式的數(shù)據(jù)加載為dataframe。Libsvm是由國立臺灣大學(xué)以C++語言開發(fā)的一個開源機(jī)器學(xué)習(xí)庫，主要提供有關(guān)支持向量機(jī)的算法。該庫的數(shù)據(jù)格式是每一行代表一個稀疏的特征向量，格式為：<label><index1>:<value1><index2>:<value2>，…。其中<label>是數(shù)據(jù)樣本的類別，<index>是索引值，<value>是對應(yīng)位置的特征的屬性值。注意到特征向量是稀疏的，所以其他索引值對應(yīng)的屬性值默認(rèn)為0。轉(zhuǎn)換后的DataFrame有兩列：標(biāo)簽，以Double類型存儲；特征，以Vector類型存儲。LIBSVM數(shù)據(jù)源9.3.4數(shù)據(jù)源scala>valdf=spark.read.format("libsvm").|option("numFeatures","780").|load("file:///usr/local/spark/data/mllib/sample_libsvm_data.txt")

第1步：讀取Spark自帶的libsvm格式的輸入數(shù)據(jù)，實例如下LibSVMDataSource的option參數(shù)含義如下表參數(shù)含義numFeatures特征的數(shù)目。如果該參數(shù)未指定或參數(shù)值非正數(shù)，方法能夠自動確定特征的數(shù)量，但會帶來額外的開銷vectorType特征向量的類型，取值為”sparse”或“dense”，默認(rèn)值為“sparse”9.3.4數(shù)據(jù)源scala>df.show(10)+-----+--------------------+|label|features|+-----+--------------------+|0.0|(780,[127,128,129...||1.0|(780,[158,159,160...||1.0|(780,[124,125,126...||1.0|(780,[152,153,154...||1.0|(780,[151,152,153...||0.0|(780,[129,130,131...||1.0|(780,[158,159,160...||1.0|(780,[99,100,101,...||0.0|(780,[154,155,156...||0.0|(780,[127,128,129...|+-----+--------------------+onlyshowingtop10rows

第2步：輸出結(jié)果如下

基本統(tǒng)計工具9.4基本統(tǒng)計工具相關(guān)性匯總統(tǒng)計假設(shè)檢驗0102039.4.1相關(guān)性計算兩組數(shù)據(jù)之間的相關(guān)性是統(tǒng)計中的常見操作。spark.ml庫提供了在多組數(shù)據(jù)之間計算兩兩相關(guān)性的方法。目前支持的相關(guān)性方法有皮爾森相關(guān)和斯皮爾曼相關(guān)，其相關(guān)系數(shù)可以反應(yīng)兩個變量之間變化趨勢的方向以及程度。皮爾森相關(guān)系數(shù)是一種線性相關(guān)系數(shù)，如上：皮爾森相關(guān)系數(shù)的輸出范圍介于-1到+1之間，0表示無相關(guān)性，正值表示正相關(guān)，負(fù)值表示負(fù)相關(guān)，相關(guān)系數(shù)的絕對值越大，相關(guān)度越強(qiáng)，相關(guān)系數(shù)越接近于0，相關(guān)度越弱。斯皮爾曼相關(guān)系數(shù)：皮爾森相關(guān)系數(shù)主要用于服從正態(tài)分布的變量，對不服從正態(tài)分布的變量，可以使用斯皮爾曼相關(guān)系數(shù)進(jìn)行相關(guān)性分析，斯皮爾曼相關(guān)系數(shù)可以更好地用于測量變量的排序關(guān)系，其計算公式為：9.4.1相關(guān)性9.4.1相關(guān)性scala>importorg.apache.spark.ml.linalg.{Matrix,Vectors}scala>importorg.apache.spark.ml.stat.Correlationscala>importorg.apache.spark.sql.Row

使用spark.ml庫提供的方法進(jìn)行相關(guān)性分析的實例-第1步：導(dǎo)入相關(guān)性方法所需要的包scala>valdata=Seq(|Vectors.sparse(4,Seq((0,2.0),(2,-1.0))),|Vectors.dense(3.0,0.0,4.0,5.0),|Vectors.dense(6.0,8.0,0.0,7.0))data:Seq[org.apache.spark.ml.linalg.Vector]=List((4,[0,2],[2.0,-1.0]),[3.0,0.0,4.0,5.0],[6.0,8.0,0.0,7.0])scala>valdf=data.map(Tuple1.apply).toDF("features")df:org.apache.spark.sql.DataFrame=[features:vector]

第2步：創(chuàng)建實驗數(shù)據(jù)，并轉(zhuǎn)化成DataFrame9.4.1相關(guān)性scala>valRow(coeff1:Matrix)=Correlation.corr(df,"features").headcoeff1:org.apache.spark.ml.linalg.Matrix=1.0 0.9707253433941511 -0.090784129900320370.86602540378443870.9707253433941511 1.0 -0.32732683535398850.720576692122892-0.09078412990032037 -0.3273268353539885 1.0 0.41931393468876730.8660254037844387 0.720576692122892 0.41931393468876731.0

第3步：調(diào)用Correlation包中的corr()函數(shù)來獲取皮爾森相關(guān)性系數(shù)9.4.1相關(guān)性scala>valRow(coeff2:Matrix)=Correlation.corr(df,"features","spearman").headcoeff2:org.apache.spark.ml.linalg.Matrix=1.0 0.8660254037844387 0.5 1.00.8660254037844387 1.0 0.0 0.86602540378443870.5 0.0 1.0 0.51.0 0.8660254037844387 0.5 1.0

第4步：使用指定的斯皮爾曼相關(guān)性方法計算輸入數(shù)據(jù)集的相關(guān)性9.4.2假設(shè)檢驗適配度檢驗用于驗證一組觀察值的次數(shù)分配是否異于理論上的分配。獨立性檢驗用于驗證從兩個變量抽出的配對觀察值組是否互相獨立。Spark.ml庫目前支持皮爾森卡方檢驗的獨立性檢驗獨立性檢驗一般采用列聯(lián)表的形式記錄觀察數(shù)據(jù),列聯(lián)表是由兩個以上的變量進(jìn)行交叉分類的頻數(shù)分布表假設(shè)檢驗用來確定結(jié)果是否具有統(tǒng)計意義，以及該結(jié)果是否偶然發(fā)生。而卡方檢驗是用途較廣的一種假設(shè)檢驗方法，包括適合度檢驗和獨立性檢驗9.4.2假設(shè)檢驗獨立性檢驗的步驟為，首先需要建立原假設(shè)，即H0：兩變量相互獨立，H1：兩變量相互不獨立。接著計算自由度和卡方檢驗的統(tǒng)計值。自由度的計算公式為（假設(shè)列聯(lián)表共有r行c列）：卡方檢驗的統(tǒng)計值的計算公式為：卡方檢驗的統(tǒng)計值的計算公式為：01根據(jù)設(shè)定的置信水準(zhǔn)，查出自由度為df的卡方分配臨界值，比較它與卡方檢驗的統(tǒng)計值可以推測是否能拒絕原假設(shè)。029.4.2假設(shè)檢驗scala>importorg.apache.spark.ml.linalg.{Vector,Vectors}scala>importorg.apache.spark.ml.stat.ChiSquareTestorg.apache.spark.ml.stat.ChiSquareTest

使用spark.ml庫提供的方法進(jìn)行相關(guān)性分析的實例-第1步：導(dǎo)入卡方檢驗所需要的包9.4.2假設(shè)檢驗scala>valdata=Seq(|(0.0,Vectors.dense(3.5,40.0)),|(0.0,Vectors.dense(3.5,30.0)),|(1.0,Vectors.dense(1.5,30.0)),|(0.0,Vectors.dense(1.5,20.0)),|(0.0,Vectors.dense(0.5,10.0)))data:Seq[(Double,org.apache.spark.ml.linalg.Vector)]=List((0.0,[3.5,40.0]),(0.0,[3.5,30.0]),(1.0,[1.5,30.0]),(0.0,[1.5,20.0]),(0.0,[0.5,10.0]))scala>valdf=data.toDF("label","features")df:org.apache.spark.sql.DataFrame=[label:double,features:vector]

第2步：創(chuàng)建實驗數(shù)據(jù)，具體代碼如下9.4.2假設(shè)檢驗scala>valchi=ChiSquareTest.test(df,"features","label").headchi:org.apache.spark.sql.Row=[[0.3916056266767989,0.5987516330675617],WrappedArray(2,3),[1.8750000000000002,1.875]]

第3步：調(diào)用test()函數(shù)，將(特征，標(biāo)簽)對轉(zhuǎn)換成一個列聯(lián)矩陣，計算卡方統(tǒng)計量9.4.2假設(shè)檢驗scala>println(s"pValues=${chi.getAs[Vector](0)}")pValues=[0.3916056266767989,0.5987516330675617]scala>println(s"degreesOfFreedom${chi.getSeq[Int](1).mkString("[",",","]")}")degreesOfFreedom[2,3]scala>println(s"statistics${chi.getAs[Vector](2)}")statistics[1.8750000000000002,1.875]

第4步：分別獲取卡方分配右尾機(jī)率、自由度、統(tǒng)計值9.4.3匯總統(tǒng)計3.0Summarizer中的包最大值最小值平均值總和、方差標(biāo)準(zhǔn)差非零數(shù)總計數(shù)查看列9.4.3匯總統(tǒng)計scala>importorg.apache.spark.ml.linalg.{Vector,Vectors}scala>importorg.apache.spark.ml.stat.Summarizerscala>importspark.implicits._scala>importSummarizer._第1步：導(dǎo)入?yún)R總統(tǒng)計所需要的包9.4.3匯總統(tǒng)計scala>valdata=Seq(|(Vectors.dense(1.0,2.0,4.0),1.0),|(Vectors.dense(4.0,3.0,6.0),3.0))data:Seq[(org.apache.spark.ml.linalg.Vector,Double)]=List(([1.0,2.0,4.0],1.0),([4.0,3.0,6.0],3.0))scala>valdf=data.toDF("features","weight")df:org.apache.spark.sql.DataFrame=[features:vector,weight:double]

第2步：創(chuàng)建實驗數(shù)據(jù)，具體代碼如下9.4.3匯總統(tǒng)計scala>val(meanVal,varianceVal)=df.select(metrics("mean","variance").|summary($"features",$"weight").as("summary")).|select("summary.mean","summary.variance").|as[(Vector,Vector)].first()meanVal:org.apache.spark.ml.linalg.Vector=[3.25,2.75,5.5]varianceVal:org.apache.spark.ml.linalg.Vector=[4.5,0.5,2.0]

第3步：計算得到數(shù)據(jù)的加權(quán)平均值和加權(quán)方差9.4.3匯總統(tǒng)計scala>val(meanVal2,varianceVal2)=df.select(mean($"features"),variance($"features")).|as[(Vector,Vector)].first()meanVal2:org.apache.spark.ml.linalg.Vector=[2.5,2.5,5.0]varianceVal2:org.apache.spark.ml.linalg.Vector=[4.5,0.5,2.0]

第4步：計算得到數(shù)據(jù)無權(quán)重下的平均值和方差

機(jī)器學(xué)習(xí)流水線9.5機(jī)器學(xué)習(xí)流水線9.5.1流水線的概念9.5.2流水線工作過程9.5.1流水線的概念01DataFrame它被MLPipeline用來存儲源數(shù)據(jù)。例如，DataFrame中的列可以是存儲的文本、特征向量、真實標(biāo)簽和預(yù)測的標(biāo)簽等02MLPipeline使用SparkSQL中的DataFrame作為數(shù)據(jù)集，它可以容納各種數(shù)據(jù)類型。較之RDD，DataFrame包含了schema信息，更類似傳統(tǒng)數(shù)據(jù)庫中的二維表格9.5.1流水線的概念Transformer轉(zhuǎn)換器DataFrameDataFrame比如一個模型就是一個Transformer。它可把一個不包含預(yù)測標(biāo)簽的測試數(shù)據(jù)集DataFrame打上標(biāo)簽，轉(zhuǎn)化成另一個包含預(yù)測標(biāo)簽的DataFrame。技術(shù)上Transformer實現(xiàn)了一個方法transform()，它通過附加一個或多個列將一個DataFrame轉(zhuǎn)換為另一個DataFrame9.5.1流水線的概念估計器或評估器Estimator翻譯成估計器或評估器，它是學(xué)習(xí)算法或在訓(xùn)練數(shù)據(jù)上的訓(xùn)練方法的概念抽象。在Pipeline里通常是被用來操作DataFrame數(shù)據(jù)并生成一個Transformer。從技術(shù)上講，Estimator實現(xiàn)了一個方法fit()，它接受一個DataFrame并產(chǎn)生一個轉(zhuǎn)換器。比如，一個隨機(jī)森林算法就是一個Estimator，它可以調(diào)用fit()，通過訓(xùn)練特征數(shù)據(jù)而得到一個隨機(jī)森林模型。9.5.1流水線的概念評估器轉(zhuǎn)換器完成機(jī)器學(xué)習(xí)的功能參數(shù)設(shè)置ParameterParameter被用來設(shè)置Transformer或者Estimator的參數(shù)。現(xiàn)在，所有轉(zhuǎn)換器和估計器可共享用于指定參數(shù)的公共API。ParamMap是一組（參數(shù)，值）對9.5.1流水線的概念PipeLine機(jī)器學(xué)習(xí)流水線數(shù)據(jù)轉(zhuǎn)換器估計器流水線9.5.2流水線工作過程定義Pipeline中的各個流水線階段PipelineStage構(gòu)建Pipeline流水線包括轉(zhuǎn)換器評估器轉(zhuǎn)換器和評估器有序地組織起來構(gòu)建成PipeLine按照處理邏輯構(gòu)建PipeLinevalpipeline=newPipeline().setStages(Array(stage1,stage2,stage3,…))返回一個PipelineModel類實例把訓(xùn)練數(shù)據(jù)集作為輸入?yún)?shù)調(diào)用fit()方法被用來預(yù)測測試數(shù)據(jù)的標(biāo)簽9.5.2流水線工作過程LogisticRegressionModelHashingTFTokenizerPipeline(Estimator)Pipeline.fit()RawtextWordsFeaturevectors流水線各階段運行,輸入的DataFrame在它通過每個階段時被轉(zhuǎn)換LogisticRegression9.5.2流水線工作過程值得注意的是，流水線本身也可以看做是一個估計器。在流水線的fit（）方法運行之后，它產(chǎn)生一個PipelineModel，它是一個Transformer。這個管道模型將在測試數(shù)據(jù)的時候使用HashingTFTokenizerPipelineModel(Transformer)LogisticRegressionRawtextWordsFeaturevectorsPredictionsPipelineModel.Transformer()9.5.2流水線工作過程9.5.2流水線工作過程流水線本身就是一個評估器，因此，在流水線的fit()方法運行之后，會產(chǎn)生一個流水線模型（PipelineModel），這是一個轉(zhuǎn)換器，可在測試數(shù)據(jù)的時候使用9.5.2流水線工作過程設(shè)置實例的參數(shù)例:lr是的一個LogisticRegression實例，用lr.setMaxIter(10)進(jìn)行參數(shù)設(shè)置以后，可以使lr.fit()至多迭代10次傳遞ParamMap給fit()或transform()函數(shù)ParamMap中的任何參數(shù)，將覆蓋先前通過set方法指定的參數(shù)0102將參數(shù)傳遞給算法主要有以下兩種方法MLlib特征抽取轉(zhuǎn)換和選擇9.6特征提取、轉(zhuǎn)換和選擇特征提取局部敏感哈希0102特征轉(zhuǎn)換03特征選擇049.6.1特征提取特征提取操作TF-IDFWord2VecCountVectorizerFeatureHasher01030402spark.ml包提供的提取操作包括9.6.1特征提?。═F-IDF）文本挖掘詞頻－逆向文件頻率TF-IDF詞語由t表示，文檔由d表示，語料庫由D表示。詞頻TF(t,d)是詞語t在文檔d中出現(xiàn)的次數(shù)。文件頻率DF(t,D)是包含詞語的文檔的個數(shù)TF-IDF

度量值表示（TF-IDF）9.6.1特征提取TF-IDFTFIDF評估器TF:

HashingTF是一個Transformer，在文本處理中，接收詞條的集合然后把這些集合轉(zhuǎn)化成固定長度的特征向量。這個算法在哈希的同時會統(tǒng)計各個詞條的詞頻IDF:

IDF是一個Estimator，在一個數(shù)據(jù)集上應(yīng)用它的fit()方法，產(chǎn)生一個IDFModel。該IDFModel接收特征向量（由HashingTF產(chǎn)生），然后計算每一個詞在文檔中出現(xiàn)的頻次。IDF會減少那些在語料庫中出現(xiàn)頻率較高的詞的權(quán)重9.6.1特征提取在下面的代碼段中，我們以一組句子開始過程描述首先使用分解器Tokenizer把句子劃分為單個詞語對每一個句子（詞袋），使用HashingTF將句子轉(zhuǎn)換為特征向量最后使用IDF重新調(diào)整特征向量9.6.1特征提取importorg.apache.spark.ml.feature.{HashingTF,IDF,Tokenizer}

第1步：導(dǎo)入TF-IDF所需要的包scala>valsentenceData=spark.createDataFrame(Seq(|(0,"IheardaboutSparkandIloveSpark"),|(0,"IwishJavacouldusecaseclasses"),|(1,"Logisticregressionmodelsareneat")|)).toDF("label","sentence")sentenceData:org.apache.spark.sql.DataFrame=[label:int,sentence:string]

第2步：創(chuàng)建一個集合，每一個句子代表一個文件scala>valtokenizer=newTokenizer().setInputCol("sentence").setOutputCol("words")tokenizer:org.apache.spark.ml.feature.Tokenizer=tok_494411a37f99

scala>valwordsData=tokenizer.transform(sentenceData)wordsData:org.apache.spark.sql.DataFrame=[label:int,sentence:string,words:array<string>]

scala>wordsData.show(false)+-----+------------------------------------+---------------------------------------------+|label|sentence|words|+-----+------------------------------------+---------------------------------------------+|0|IheardaboutSparkandIloveSpark|[i,heard,about,spark,and,i,love,spark]||0|IwishJavacouldusecaseclasses|[i,wish,java,could,use,case,classes]||1|Logisticregressionmodelsareneat|[logistic,regression,models,are,neat]|+-----+------------------------------------+---------------------------------------------+

第3步：用Tokenizer把每個句子分解成單詞9.6.1特征提取9.6.1特征提取scala>valhashingTF=newHashingTF().|setInputCol("words").setOutputCol("rawFeatures").setNumFeatures(2000)hashingTF:org.apache.spark.ml.feature.HashingTF=hashingTF_2591ec73cea0

scala>valfeaturizedData=hashingTF.transform(wordsData)featurizedData:org.apache.spark.sql.DataFrame=[label:int,sentence:string,words:array<string>,rawFeatures:vector]

第4步：用HashingTF的transform()方法把每個“詞袋”哈希成特征向量9.6.1特征提取scala>featurizedData.select("words","rawFeatures").show(false)+---------------------------------------------+---------------------------------------------------------------------+|words|rawFeatures|+---------------------------------------------+---------------------------------------------------------------------+|[i,heard,about,spark,and,i,love,spark]|(2000,[240,333,1105,1329,1357,1777],[1.0,1.0,2.0,2.0,1.0,1.0])||[i,wish,java,could,use,case,classes]|(2000,[213,342,489,495,1329,1809,1967],[1.0,1.0,1.0,1.0,1.0,1.0,1.0])||[logistic,regression,models,are,neat]|(2000,[286,695,1138,1193,1604],[1.0,1.0,1.0,1.0,1.0])|+---------------------------------------------+---------------------------------------------------------------------+

第4步：用HashingTF的transform()方法把每個“詞袋”哈希成特征向量9.6.1特征提取scala>validf=newIDF().setInputCol("rawFeatures").setOutputCol("features")idf:org.apache.spark.ml.feature.IDF=idf_7fcc9063de6f

scala>validfModel=idf.fit(featurizedData)idfModel:org.apache.spark.ml.feature.IDFModel=idf_7fcc9063de6f

第5步：調(diào)用IDF方法來重新構(gòu)造特征向量的規(guī)模，生成的變量idf是一個評估器9.6.1特征提取scala>valrescaledData=idfModel.transform(featurizedData)rescaledData:org.apache.spark.sql.DataFrame=[label:int,sentence:string,words:array<string>,rawFeatures:vector,features:vector]

第6步：調(diào)用IDFModel的transform()方法，可以得到每一個單詞對應(yīng)的TF-IDF度量值9.6.1特征提取scala>rescaledData.select("features","label").show(false)+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-----+|features|label|+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-----+|(2000,[240,333,1105,1329,1357,1777],[0.6931471805599453,0.6931471805599453,1.3862943611198906,0.5753641449035617,0.6931471805599453,0.6931471805599453])|0||(2000,[213,342,489,495,1329,1809,1967],[0.6931471805599453,0.6931471805599453,0.6931471805599453,0.6931471805599453,0.28768207245178085,0.6931471805599453,0.6931471805599453])|0||(2000,[286,695,1138,1193,1604],[0.6931471805599453,0.6931471805599453,0.6931471805599453,0.6931471805599453,0.6931471805599453])|1|+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-----+

第6步：調(diào)用IDFModel的transform()方法，可以得到每一個單詞對應(yīng)的TF-IDF度量值9.6.1特征提取name:="SimpleProject"version:="1.0"scalaVersion:="2.12.15"libraryDependencies+="org.apache.spark"%"spark-mllib_2.12"%"3.2.0"simple.sbt文件為了方便調(diào)試和觀察執(zhí)行效果，本章的代碼都是在spark-shell中執(zhí)行，實際上，也可編寫?yīng)毩?yīng)用程序，用sbt編譯打包后使用spark-submit命令運行，具體方法同前幾章需要注意的是9.6.2特征轉(zhuǎn)換（1）Tokenizer（2）StopWordsRemover（3）NGram（4）Binarizer（5）PCA（6）PolynomialExpansion（7）DiscreteCosineTransform(DCT)（8）StringIndexer（9）IndexToString（10）OneHotEncoder（11）VectorIndexer（12）Interaction（13）Normalizer（14）StandardScaler（15）RobustScaler（16）MinMaxScaler（17）MaxAbsScaler（18）Bucketizer（19）ElementwiseProduct（20）SQLTransformer（21）VectorAssembler（22）VectorSizeHint（23）QuantileDiscretizer（24）Imputerspark.ml包提供了大量的特征轉(zhuǎn)換操作：9.6.2特征轉(zhuǎn)換在機(jī)器學(xué)習(xí)處理過程中，為了方便相關(guān)算法的實現(xiàn)，經(jīng)常需要把標(biāo)簽數(shù)據(jù)（一般是字符串）轉(zhuǎn)化成整數(shù)索引，或是在計算結(jié)束后將整數(shù)索引還原為相應(yīng)的標(biāo)簽SparkML包中提供了幾個相關(guān)的轉(zhuǎn)換器，例如：StringIndexer、IndexToString、OneHotEncoder、VectorIndexer，它們提供了十分方便的特征轉(zhuǎn)換功能，這些轉(zhuǎn)換器類都位于org.apache.spark.ml.feature包下Word2Vec9.6.2特征轉(zhuǎn)換StringIndexerStringIndexer轉(zhuǎn)換器可以把一列類別型的特征（或標(biāo)簽）進(jìn)行編碼，使其數(shù)值化，索引的范圍從0開始，該過程可以使得相應(yīng)的特征索引化，使得某些無法接受類別型特征的算法可以使用，并提高諸如決策樹等機(jī)器學(xué)習(xí)算法的效率如果輸入數(shù)值型的，會先把它轉(zhuǎn)化成字符型，再對其進(jìn)行編碼索引構(gòu)建的順序為標(biāo)簽的頻率，優(yōu)先編碼頻率較大的標(biāo)簽，所以出現(xiàn)頻率最高的標(biāo)簽為0號9.6.2特征轉(zhuǎn)換scala>importorg.apache.spark.ml.feature.StringIndexer

首先，引入所需要使用的類，具體代碼如下importorg.apache.spark.ml.feature.{StringIndexer,StringIndexerModel}

scala>valdf1=spark.createDataFrame(Seq(|(0,"a"),|(1,"b"),|(2,"c"),|(3,"a"),|(4,"a"),|(5,"c"))).toDF("id","category")df1:org.apache.spark.sql.DataFrame=[id:int,category:string]

其次，構(gòu)建1個DataFrame，設(shè)置StringIndexer的輸入列和輸出列的名字9.6.2特征轉(zhuǎn)換scala>valindexer=newStringIndexer().|setInputCol("category").|setOutputCol("categoryIndex")

創(chuàng)建一個StringIndexer對象，設(shè)定輸入輸出列名，其余參數(shù)采用默認(rèn)值9.6.2特征轉(zhuǎn)換scala>valindexed1=indexer.fit(df1).transform(df1)indexed1:org.apache.spark.sql.DataFrame=[id:int,category:string,categoryIndex:double]scala>indexed1.show()+---+--------+-------------+|id|category|categoryIndex|+---+--------+-------------+|0|a|0.0||1|b|2.0||2|c|1.0||3|a|0.0||4|a|0.0||5|c|1.0|+---+--------+-------------+

通過fit()方法進(jìn)行模型訓(xùn)練，用訓(xùn)練的模型對原數(shù)據(jù)集進(jìn)行處理，indexed1.show()展示9.6.2特征轉(zhuǎn)換IndexToString與StringIndexer相對應(yīng)，IndexToString的作用是把標(biāo)簽索引的一列重新映射回原有的字符型標(biāo)簽其主要使用場景一般都是和StringIndexer配合，先用StringIndexer將標(biāo)簽轉(zhuǎn)化成標(biāo)簽索引，進(jìn)行模型訓(xùn)練，然后在預(yù)測標(biāo)簽的時候再把標(biāo)簽索引轉(zhuǎn)化成原有的字符標(biāo)簽scala>importorg.apache.spark.ml.feature.IndexToStringscala>validx2str=newIndexToString().|setInputCol("categoryIndex").|setOutputCol("originalCategory")scala>valindexString=idx2str.transform(indexed1)indexString:org.apache.spark.sql.DataFrame=[id:int,category:string,categoryIndex:double,originalCategory:string]scala>indexString.select("id","originalCategory").show()+---+----------------+|id|originalCategory|+---+----------------+|0|a||1|b||2|c||3|a||4|a||5|c|+---+----------------+9.6.2特征轉(zhuǎn)換VectorIndexer之前介紹的StringIndexer是針對單個類別型特征進(jìn)行轉(zhuǎn)換，倘若所有特征都已經(jīng)被組織在一個向量中，又想對其中某些單個分量進(jìn)行處理時，SparkML提供了VectorIndexer類來解決向量數(shù)據(jù)集中的類別性特征轉(zhuǎn)換通過為其提供maxCategories超參數(shù)，它可以自動識別哪些特征是類別型的，并且將原始值轉(zhuǎn)換為類別索引。它基于不同特征值的數(shù)量來識別哪些特征需要被類別化，那些取值可能性最多不超過maxCategories的特征需要會被認(rèn)為是類別型的9.6.2特征轉(zhuǎn)換scala>importorg.apache.spark.ml.feature.VectorIndexerscala>importorg.apache.spark.ml.linalg.{Vector,Vectors}scala>valdata=Seq(|

Vectors.dense(-1.0,1.0,1.0),|

Vectors.dense(-1.0,3.0,1.0),|

Vectors.dense(0.0,5.0,1.0))data:Seq[org.apache.spark.ml.linalg.Vector]=List([-1.0,1.0,1.0],[-1.0,3.0,1.0],[0.0,5.0,1.0])scala>valdf=spark.createDataFrame(data.map(Tuple1.apply)).toDF("features")df:org.apache.spark.sql.DataFrame=[features:vector]

首先引入所需要的類，并構(gòu)建數(shù)據(jù)集，具體代碼如下9.6.2特征轉(zhuǎn)換scala>valindexer=newVectorIndexer().|

setInputCol("features").|

setOutputCol("indexed").|

setMaxCategories(2)scala>valindexerModel=indexer.fit(df)

然后，構(gòu)建VectorIndexer轉(zhuǎn)換器，設(shè)置輸入和輸出列，并進(jìn)行模型訓(xùn)練9.6.2特征轉(zhuǎn)換設(shè)置maxCategories為2，即只有種類小于2的特征才被認(rèn)為是類別型特征，否則被認(rèn)為是連續(xù)型特征scala>valcategoricalFeatures:Set[Int]=indexerModel.categoryMaps.keys.toSetcategoricalFeatures:Set[Int]=Set(0,2)scala>println(s"Chose${categoricalFeatures.size}categoricalfeatures:"+categoricalFeatures.mkString(","))Chose2categoricalfeatures:0,2

通過VectorIndexerModel的categoryMaps成員來獲得被轉(zhuǎn)換的特征及其映射9.6.2特征轉(zhuǎn)換scala>valindexed=indexerModel.transform(df)indexed:org.apache.spark.sql.DataFrame=[features:vector,indexed:vector]scala>indexed.show()+--------------+-------------+|

features|

indexed|+--------------+-------------+|[-1.0,1.0,1.0]|[1.0,1.0,0.0]||[-1.0,3.0,1.0]|[1.0,3.0,0.0]||[0.0,5.0,1.0]|[0.0,5.0,0.0]|+--------------+-------------+

最后，把模型應(yīng)用于原有的數(shù)據(jù)，并打印結(jié)果，具體代碼如下9.6.2特征轉(zhuǎn)換9.6.3特征選擇特征選擇是指在特征向量中選擇出那些“優(yōu)秀”的特征，組成新的、更“精簡”的特征向量的過程。它在高維數(shù)據(jù)分析中十分常用，可以剔除掉“冗余”和“無關(guān)”的特征，提升學(xué)習(xí)器的性能9.6.3特征選擇VectorSlicerVectorSlicer的作用類似于MATLAB/numpy中的“列切片”，它可以根據(jù)給定的索引（整數(shù)索引值或列名索引）選擇出特征向量中的部分列，并生成新的特征向量RFormulaRFormula提供一種R語言風(fēng)格的特征向量列選擇功能，用戶可以給其傳入一個R表達(dá)式，它會根據(jù)表達(dá)式，自動選擇相應(yīng)的特征列形成新的特征向量UnivariateFeatureSelector可看作ChiSqSelector的擴(kuò)展版，它支持對連續(xù)型數(shù)據(jù)的特征選擇，Spark自動根據(jù)特征類型和標(biāo)簽類型來選擇評估函數(shù)VarianceThresholdSelector將方差值小于人為給定的閾值的特征刪除。該閾值默認(rèn)為0，只有方差為0的特征，即所有值都相同的特征，會被刪除ChiSqSelector通過卡方選擇的方法來進(jìn)行特征選擇，它的輸入需要是一個已有標(biāo)簽的數(shù)據(jù)集9.6.3特征選擇卡方選擇是統(tǒng)計學(xué)上常用的一種有監(jiān)督特征選擇方法，它通過對特征和真實標(biāo)簽之間進(jìn)行卡方檢驗，來判斷該特征和真實標(biāo)簽的關(guān)聯(lián)程度，進(jìn)而確定是否對其進(jìn)行選擇和spark.ml包中的大多數(shù)學(xué)習(xí)方法一樣，spark.ml包中的卡方選擇也是以“評估器+轉(zhuǎn)換器”的形式出現(xiàn)，主要是由ChiSqSelector和ChiSqSelectorModel兩個類實現(xiàn)01029.6.3特征選擇scala>importorg.apache.spark.ml.feature.{ChiSqSelector,ChiSqSelectorModel}scala>importorg.apache.spark.ml.linalg.Vectors

首先，進(jìn)行環(huán)境的設(shè)置，引入卡方選擇器所需要使用的類，具體代碼如下9.6.3特征選擇scala>valdf=spark.createDataFrame(Seq(|(1,Vectors.dense(0.0,0.0,18.0,1.0),1),|(2,Vectors.dense(0.0,1.0,12.0,0.0),0),|(3,Vectors.dense(1.0,0.0,15.0,0.1),0)|)).toDF("id","features","label")df:org.apache.spark.sql.DataFrame=[id:int,features:vector...1morefield]scala>df.show()+---+------------------+-----+|id|features|label|+---+------------------+-----+|1|[0.0,0.0,18.0,1.0]|1||2|[0.0,1.0,12.0,0.0]|0||3|[1.0,0.0,15.0,0.1]|0|+---+------------------+-----+

其次，創(chuàng)建實驗數(shù)據(jù)，這是一個具有3個樣本、4個特征維度的數(shù)據(jù)集，標(biāo)簽有1和0兩種9.6.3特征選擇scala>valselector=newChiSqSelector().|setNumTopFeatures(1).|setFeaturesCol("features").|setLabelCol("label").|setOutputCol("selected-feature")

用卡方選擇進(jìn)行特征選擇器的訓(xùn)練，為了便于觀察設(shè)置只選擇和標(biāo)簽關(guān)聯(lián)性最強(qiáng)的一個特征9.6.3特征選擇scala>valselector_model=selector.fit(df)scala>valresult=selector_model.transform(df)result:org.apache.spark.sql.DataFrame=[id:int,features:vector...2morefields]scala>result.show(false)+---+------------------+-----+----------------+|id|features|label|selected-feature|+---+------------------+-----+----------------+|1|[0.0,0.0,18.0,1.0]|1.0|[18.0]||2|[0.0,1.0,12.0,0.0]|0.0|[12.0]||3|[1.0,0.0,15.0,0.1]|0.0|[15.0