提綱8.1SparkMLlib簡(jiǎn)介8.2機(jī)器學(xué)習(xí)工作流8.3特征抽取、轉(zhuǎn)化和選擇8.4分類(lèi)與回歸8.1SparkMLlib簡(jiǎn)介8.1.1什么是機(jī)器學(xué)習(xí)8.1.2基于大數(shù)據(jù)的機(jī)器學(xué)習(xí)8.1.3Spark機(jī)器學(xué)習(xí)庫(kù)MLLib8.1.1什么是機(jī)器學(xué)習(xí)機(jī)器學(xué)習(xí)可以看做是一門(mén)人工智能的科學(xué)，該領(lǐng)域的主要研究對(duì)象是人工智能。機(jī)器學(xué)習(xí)利用數(shù)據(jù)或以往的經(jīng)驗(yàn)，以此優(yōu)化計(jì)算機(jī)程序的性能標(biāo)準(zhǔn)。機(jī)器學(xué)習(xí)強(qiáng)調(diào)三個(gè)關(guān)鍵詞：算法、經(jīng)驗(yàn)、性能8.1.2基于大數(shù)據(jù)的機(jī)器學(xué)習(xí)傳統(tǒng)的機(jī)器學(xué)習(xí)算法，由于技術(shù)和單機(jī)存儲(chǔ)的限制，只能在少量數(shù)據(jù)上使用，依賴(lài)于數(shù)據(jù)抽樣大數(shù)據(jù)技術(shù)的出現(xiàn)，可以支持在全量數(shù)據(jù)上進(jìn)行機(jī)器學(xué)習(xí)機(jī)器學(xué)習(xí)算法涉及大量迭代計(jì)算基于磁盤(pán)的MapReduce不適合進(jìn)行大量迭代計(jì)算基于內(nèi)存的Spark比較適合進(jìn)行大量迭代計(jì)算8.1.3Spark機(jī)器學(xué)習(xí)庫(kù)MLlibSpark提供了一個(gè)基于海量數(shù)據(jù)的機(jī)器學(xué)習(xí)庫(kù)，它提供了常用機(jī)器學(xué)習(xí)算法的分布式實(shí)現(xiàn)開(kāi)發(fā)者只需要有Spark基礎(chǔ)并且了解機(jī)器學(xué)習(xí)算法的原理，以及方法相關(guān)參數(shù)的含義，就可以輕松的通過(guò)調(diào)用相應(yīng)的API來(lái)實(shí)現(xiàn)基于海量數(shù)據(jù)的機(jī)器學(xué)習(xí)過(guò)程pyspark的即席查詢(xún)也是一個(gè)關(guān)鍵。算法工程師可以邊寫(xiě)代碼邊運(yùn)行，邊看結(jié)果8.1.3Spark機(jī)器學(xué)習(xí)庫(kù)MLlib需要注意的是，MLlib中只包含能夠在集群上運(yùn)行良好的并行算法，這一點(diǎn)很重要有些經(jīng)典的機(jī)器學(xué)習(xí)算法沒(méi)有包含在其中，就是因?yàn)樗鼈儾荒懿⑿袌?zhí)行相反地，一些較新的研究得出的算法因?yàn)檫m用于集群，也被包含在MLlib中，例如分布式隨機(jī)森林算法、最小交替二乘算法。這樣的選擇使得MLlib中的每一個(gè)算法都適用于大規(guī)模數(shù)據(jù)集如果是小規(guī)模數(shù)據(jù)集上訓(xùn)練各機(jī)器學(xué)習(xí)模型，最好還是在各個(gè)節(jié)點(diǎn)上使用單節(jié)點(diǎn)的機(jī)器學(xué)習(xí)算法庫(kù)（比如Weka）8.1.3Spark機(jī)器學(xué)習(xí)庫(kù)MLlibMLlib是Spark的機(jī)器學(xué)習(xí)（MachineLearning）庫(kù)，旨在簡(jiǎn)化機(jī)器學(xué)習(xí)的工程實(shí)踐工作MLlib由一些通用的學(xué)習(xí)算法和工具組成，包括分類(lèi)、回歸、聚類(lèi)、協(xié)同過(guò)濾、降維等，同時(shí)還包括底層的優(yōu)化原語(yǔ)和高層的流水線（Pipeline）API，具體如下：算法工具：常用的學(xué)習(xí)算法，如分類(lèi)、回歸、聚類(lèi)和協(xié)同過(guò)濾；特征化工具：特征提取、轉(zhuǎn)化、降維和選擇工具；流水線(Pipeline)：用于構(gòu)建、評(píng)估和調(diào)整機(jī)器學(xué)習(xí)工作流的工具;持久性：保存和加載算法、模型和管道;實(shí)用工具：線性代數(shù)、統(tǒng)計(jì)、數(shù)據(jù)處理等工具。8.1.3Spark機(jī)器學(xué)習(xí)庫(kù)MLlibSpark機(jī)器學(xué)習(xí)庫(kù)從1.2版本以后被分為兩個(gè)包：spark.mllib

包含基于RDD的原始算法API。SparkMLlib歷史比較長(zhǎng)，在1.0以前的版本即已經(jīng)包含了，提供的算法實(shí)現(xiàn)都是基于原始的RDDspark.ml則提供了基于DataFrames高層次的API，可以用來(lái)構(gòu)建機(jī)器學(xué)習(xí)工作流（PipeLine）。MLPipeline彌補(bǔ)了原始MLlib庫(kù)的不足，向用戶(hù)提供了一個(gè)基于DataFrame的機(jī)器學(xué)習(xí)工作流式API套件8.1.3Spark機(jī)器學(xué)習(xí)庫(kù)MLlibMLlib目前支持4種常見(jiàn)的機(jī)器學(xué)習(xí)問(wèn)題:

分類(lèi)、回歸、聚類(lèi)和協(xié)同過(guò)濾8.2機(jī)器學(xué)習(xí)流水線8.2.1機(jī)器學(xué)習(xí)流水線概念8.2.2構(gòu)建一個(gè)機(jī)器學(xué)習(xí)流水線8.2.1機(jī)器學(xué)習(xí)流水線概念在介紹流水線之前，先來(lái)了解幾個(gè)重要概念：DataFrame：使用SparkSQL中的DataFrame作為數(shù)據(jù)集，它可以容納各種數(shù)據(jù)類(lèi)型。較之RDD，DataFrame包含了schema信息，更類(lèi)似傳統(tǒng)數(shù)據(jù)庫(kù)中的二維表格。它被MLPipeline用來(lái)存儲(chǔ)源數(shù)據(jù)。例如，DataFrame中的列可以是存儲(chǔ)的文本、特征向量、真實(shí)標(biāo)簽和預(yù)測(cè)的標(biāo)簽等8.2.1機(jī)器學(xué)習(xí)流水線概念Transformer：翻譯成轉(zhuǎn)換器，是一種可以將一個(gè)DataFrame轉(zhuǎn)換為另一個(gè)DataFrame的算法。比如一個(gè)模型就是一個(gè)Transformer。它可以把一個(gè)不包含預(yù)測(cè)標(biāo)簽的測(cè)試數(shù)據(jù)集DataFrame打上標(biāo)簽，轉(zhuǎn)化成另一個(gè)包含預(yù)測(cè)標(biāo)簽的DataFrame。技術(shù)上，Transformer實(shí)現(xiàn)了一個(gè)方法transform()，它通過(guò)附加一個(gè)或多個(gè)列將一個(gè)DataFrame轉(zhuǎn)換為另一個(gè)DataFrame8.2.1機(jī)器學(xué)習(xí)流水線概念Estimator：翻譯成估計(jì)器或評(píng)估器，它是學(xué)習(xí)算法或在訓(xùn)練數(shù)據(jù)上的訓(xùn)練方法的概念抽象。在Pipeline里通常是被用來(lái)操作DataFrame數(shù)據(jù)并生成一個(gè)Transformer。從技術(shù)上講，Estimator實(shí)現(xiàn)了一個(gè)方法fit()，它接受一個(gè)DataFrame并產(chǎn)生一個(gè)轉(zhuǎn)換器。比如，一個(gè)隨機(jī)森林算法就是一個(gè)Estimator，它可以調(diào)用fit()，通過(guò)訓(xùn)練特征數(shù)據(jù)而得到一個(gè)隨機(jī)森林模型。8.2.1機(jī)器學(xué)習(xí)流水線概念Parameter：Parameter被用來(lái)設(shè)置Transformer或者Estimator的參數(shù)?，F(xiàn)在，所有轉(zhuǎn)換器和估計(jì)器可共享用于指定參數(shù)的公共API。ParamMap是一組（參數(shù)，值）對(duì)PipeLine：翻譯為流水線或者管道。流水線將多個(gè)工作流階段（轉(zhuǎn)換器和估計(jì)器）連接在一起，形成機(jī)器學(xué)習(xí)的工作流，并獲得結(jié)果輸出8.2.2流水線工作過(guò)程要構(gòu)建一個(gè)Pipeline流水線，首先需要定義Pipeline中的各個(gè)流水線階段PipelineStage（包括轉(zhuǎn)換器和評(píng)估器），比如指標(biāo)提取和轉(zhuǎn)換模型訓(xùn)練等。有了這些處理特定問(wèn)題的轉(zhuǎn)換器和評(píng)估器，就可以按照具體的處理邏輯有序地組織PipelineStages并創(chuàng)建一個(gè)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í)例，進(jìn)而被用來(lái)預(yù)測(cè)測(cè)試數(shù)據(jù)的標(biāo)簽8.2.2流水線工作過(guò)程流水線的各個(gè)階段按順序運(yùn)行，輸入的DataFrame在它通過(guò)每個(gè)階段時(shí)被轉(zhuǎn)換8.2.2流水線工作過(guò)程值得注意的是，流水線本身也可以看做是一個(gè)估計(jì)器。在流水線的fit（）方法運(yùn)行之后，它產(chǎn)生一個(gè)PipelineModel，它是一個(gè)Transformer。這個(gè)管道模型將在測(cè)試數(shù)據(jù)的時(shí)候使用。下圖說(shuō)明了這種用法。8.2.3構(gòu)建一個(gè)機(jī)器學(xué)習(xí)流水線任務(wù)描述查找出所有包含"spark"的句子，即將包含"spark"的句子的標(biāo)簽設(shè)為1，沒(méi)有"spark"的句子的標(biāo)簽設(shè)為0。本節(jié)以邏輯斯蒂回歸為例，構(gòu)建一個(gè)典型的機(jī)器學(xué)習(xí)過(guò)程，來(lái)具體介紹一下流水線是如何應(yīng)用的8.2.3構(gòu)建一個(gè)機(jī)器學(xué)習(xí)流水線需要使用SparkSession對(duì)象Spark2.0以上版本的pyspark在啟動(dòng)時(shí)會(huì)自動(dòng)創(chuàng)建一個(gè)名為spark的SparkSession對(duì)象，當(dāng)需要手工創(chuàng)建時(shí)，SparkSession可以由其伴生對(duì)象的builder()方法創(chuàng)建出來(lái)，如下代碼段所示：frompyspark.sqlimportSparkSessionspark=SparkSession.builder.master("local").appName("WordCount").getOrCreate()pyspark.ml依賴(lài)numpy包，Ubuntu自帶python3是沒(méi)有numpy的，執(zhí)行如下命令安裝：sudopip3installnumpy8.2.3構(gòu)建一個(gè)機(jī)器學(xué)習(xí)流水線（1）引入要包含的包并構(gòu)建訓(xùn)練數(shù)據(jù)集frompyspark.mlimportPipelinefrompyspark.ml.classificationimportLogisticRegressionfrompyspark.ml.featureimportHashingTF,Tokenizer

#Preparetrainingdocumentsfromalistof(id,text,label)tuples.training=spark.createDataFrame([(0,"abcdespark",1.0),(1,"bd",0.0),(2,"sparkfgh",1.0),(3,"hadoopmapreduce",0.0)],["id","text","label"])8.2.3構(gòu)建一個(gè)機(jī)器學(xué)習(xí)流水線（2）定義Pipeline中的各個(gè)流水線階段PipelineStage，包括轉(zhuǎn)換器和評(píng)估器，具體地，包含tokenizer,hashingTF和lr。tokenizer=Tokenizer(inputCol="text",outputCol="words")hashingTF=HashingTF(inputCol=tokenizer.getOutputCol(),outputCol="features")lr=LogisticRegression(maxIter=10,regParam=0.001)8.2.3構(gòu)建一個(gè)機(jī)器學(xué)習(xí)流水線（3）按照具體的處理邏輯有序地組織PipelineStages，并創(chuàng)建一個(gè)Pipeline。pipeline=Pipeline(stages=[tokenizer,hashingTF,lr])現(xiàn)在構(gòu)建的Pipeline本質(zhì)上是一個(gè)Estimator，在它的fit()方法運(yùn)行之后，它將產(chǎn)生一個(gè)PipelineModel，它是一個(gè)Transformer。model=pipeline.fit(training)可以看到，model的類(lèi)型是一個(gè)PipelineModel，這個(gè)流水線模型將在測(cè)試數(shù)據(jù)的時(shí)候使用8.2.3構(gòu)建一個(gè)機(jī)器學(xué)習(xí)流水線（4）構(gòu)建測(cè)試數(shù)據(jù)test=spark.createDataFrame([(4,"sparkijk"),(5,"lmn"),(6,"sparkhadoopspark"),(7,"apachehadoop")],["id","text"])8.2.3構(gòu)建一個(gè)機(jī)器學(xué)習(xí)流水線（5）調(diào)用之前訓(xùn)練好的PipelineModel的transform()方法，讓測(cè)試數(shù)據(jù)按順序通過(guò)擬合的流水線，生成預(yù)測(cè)結(jié)果prediction=model.transform(test)selected=prediction.select("id","text","probability","prediction")forrowinselected.collect():rid,text,prob,prediction=rowprint("(%d,%s)-->prob=%s,prediction=%f"%(rid,text,str(prob),prediction))

(4,sparkijk)-->prob=[0.155543713844,0.844456286156],prediction=1.000000(5,lmn)-->prob=[0.830707735211,0.169292264789],prediction=0.000000(6,sparkhadoopspark)-->prob=[0.0696218406195,0.93037815938],prediction=1.000000(7,apachehadoop)-->prob=[0.981518350351,0.018481649649],prediction=0.0000008.3特征提取和轉(zhuǎn)換8.3.1特征提取8.3.2特征轉(zhuǎn)換8.3.1特征提?。篢F-IDF“詞頻－逆向文件頻率”（TF-IDF）是一種在文本挖掘中廣泛使用的特征向量化方法，它可以體現(xiàn)一個(gè)文檔中詞語(yǔ)在語(yǔ)料庫(kù)中的重要程度。詞語(yǔ)由t表示，文檔由d表示，語(yǔ)料庫(kù)由D表示。詞頻TF(t,d)是詞語(yǔ)t在文檔d中出現(xiàn)的次數(shù)。文件頻率DF(t,D)是包含詞語(yǔ)的文檔的個(gè)數(shù)。TF-IDF就是在數(shù)值化文檔信息，衡量詞語(yǔ)能提供多少信息以區(qū)分文檔。其定義如下：TF-IDF度量值表示如下：8.3.1特征提?。篢F-IDF在SparkML庫(kù)中，TF-IDF被分成兩部分：TF(+hashing)

IDFTF:

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

IDF是一個(gè)Estimator，在一個(gè)數(shù)據(jù)集上應(yīng)用它的fit()方法，產(chǎn)生一個(gè)IDFModel。該IDFModel接收特征向量（由HashingTF產(chǎn)生），然后計(jì)算每一個(gè)詞在文檔中出現(xiàn)的頻次。IDF會(huì)減少那些在語(yǔ)料庫(kù)中出現(xiàn)頻率較高的詞的權(quán)重。8.3.1特征提?。篢F-IDF過(guò)程描述：在下面的代碼段中，我們以一組句子開(kāi)始首先使用分解器Tokenizer把句子劃分為單個(gè)詞語(yǔ)對(duì)每一個(gè)句子（詞袋），使用HashingTF將句子轉(zhuǎn)換為特征向量最后使用IDF重新調(diào)整特征向量（這種轉(zhuǎn)換通?？梢蕴岣呤褂梦谋咎卣鞯男阅埽?.3.1特征提?。篢F-IDF（1）導(dǎo)入TF-IDF所需要的包：>>>frompyspark.ml.featureimportHashingTF,IDF,Tokenizer（2）創(chuàng)建一個(gè)簡(jiǎn)單的DataFrame，每一個(gè)句子代表一個(gè)文檔>>>sentenceData=spark.createDataFrame([(0,"IheardaboutSparkandIloveSpark"),(0,"IwishJavacouldusecaseclasses"),(1,"Logisticregressionmodelsareneat")]).toDF("label","sentence")8.3.1特征提?。篢F-IDF（3）得到文檔集合后，即可用tokenizer對(duì)句子進(jìn)行分詞>>>tokenizer=Tokenizer(inputCol="sentence",outputCol="words")>>>wordsData=tokenizer.transform(sentenceData)>>>wordsData.show()+-----+--------------------+--------------------+|label|sentence|words|+-----+--------------------+--------------------+|0|IheardaboutSpa...|[i,heard,about,...||0|IwishJavacould...|[i,wish,java,c...||1|Logisticregressi...|[logistic,regres...|+-----+--------------------+--------------------+8.3.1特征提?。篢F-IDF（4）得到分詞后的文檔序列后，即可使用HashingTF的transform()方法把句子哈希成特征向量，這里設(shè)置哈希表的桶數(shù)為2000>>>hashingTF=HashingTF(inputCol="words",outputCol="rawFeatures",numFeatures=2000)>>>featurizedData=hashingTF.transform(wordsData)>>>featurizedData.select("words","rawFeatures").show(truncate=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])|+---------------------------------------------+---------------------------------------------------------------------+8.3.1特征提?。篢F-IDF（5）調(diào)用IDF方法來(lái)重新構(gòu)造特征向量的規(guī)模，生成的變量idf是一個(gè)評(píng)估器，在特征向量上應(yīng)用它的fit()方法，會(huì)產(chǎn)生一個(gè)IDFModel（名稱(chēng)為idfModel）。>>>idf=IDF(inputCol="rawFeatures",outputCol="features")>>>idfModel=idf.fit(featurizedData)8.3.1特征提?。篢F-IDF>>>rescaledData=idfModel.transform(featurizedData)>>>rescaledData.select("features","label").show(truncate=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()方法，可以得到每一個(gè)單詞對(duì)應(yīng)的TF-IDF度量值。8.3.2特征轉(zhuǎn)換：標(biāo)簽和索引的轉(zhuǎn)化在機(jī)器學(xué)習(xí)處理過(guò)程中，為了方便相關(guān)算法的實(shí)現(xiàn)，經(jīng)常需要把標(biāo)簽數(shù)據(jù)（一般是字符串）轉(zhuǎn)化成整數(shù)索引，或是在計(jì)算結(jié)束后將整數(shù)索引還原為相應(yīng)的標(biāo)簽SparkML包中提供了幾個(gè)相關(guān)的轉(zhuǎn)換器，例如：StringIndexer、IndexToString、OneHotEncoder、VectorIndexer，它們提供了十分方便的特征轉(zhuǎn)換功能，這些轉(zhuǎn)換器類(lèi)都位于org.apache.spark.ml.feature包下值得注意的是，用于特征轉(zhuǎn)換的轉(zhuǎn)換器和其他的機(jī)器學(xué)習(xí)算法一樣，也屬于MLPipeline模型的一部分，可以用來(lái)構(gòu)成機(jī)器學(xué)習(xí)流水線，以StringIndexer為例，其存儲(chǔ)著進(jìn)行標(biāo)簽數(shù)值化過(guò)程的相關(guān)超參數(shù)，是一個(gè)Estimator，對(duì)其調(diào)用fit(..)方法即可生成相應(yīng)的模型StringIndexerModel類(lèi)，很顯然，它存儲(chǔ)了用于DataFrame進(jìn)行相關(guān)處理的參數(shù)，是一個(gè)Transformer（其他轉(zhuǎn)換器也是同一原理）8.3.2特征轉(zhuǎn)換：標(biāo)簽和索引的轉(zhuǎn)化StringIndexerStringIndexer轉(zhuǎn)換器可以把一列類(lèi)別型的特征（或標(biāo)簽）進(jìn)行編碼，使其數(shù)值化，索引的范圍從0開(kāi)始，該過(guò)程可以使得相應(yīng)的特征索引化，使得某些無(wú)法接受類(lèi)別型特征的算法可以使用，并提高諸如決策樹(shù)等機(jī)器學(xué)習(xí)算法的效率索引構(gòu)建的順序?yàn)闃?biāo)簽的頻率，優(yōu)先編碼頻率較大的標(biāo)簽，所以出現(xiàn)頻率最高的標(biāo)簽為0號(hào)如果輸入的是數(shù)值型的，會(huì)首先把它轉(zhuǎn)化成字符型，然后再對(duì)其進(jìn)行編碼8.3.2特征轉(zhuǎn)換：標(biāo)簽和索引的轉(zhuǎn)化（1）首先，引入所需要使用的類(lèi)。>>>frompyspark.ml.featureimportStringIndexer8.3.2特征轉(zhuǎn)換：標(biāo)簽和索引的轉(zhuǎn)化（2）其次，構(gòu)建1個(gè)DataFrame，設(shè)置StringIndexer的輸入列和輸出列的名字。>>>df=spark.createDataFrame([(0,"a"),(1,"b"),(2,"c"),(3,"a"),(4,"a"),(5,"c")],["id","category"])>>>indexer=StringIndexer(inputCol="category",outputCol="categoryIndex")8.3.2特征轉(zhuǎn)換：標(biāo)簽和索引的轉(zhuǎn)化（3）然后，通過(guò)fit()方法進(jìn)行模型訓(xùn)練，用訓(xùn)練出的模型對(duì)原數(shù)據(jù)集進(jìn)行處理，并通過(guò)indexed.show()進(jìn)行展示。>>>model=indexer.fit(df)>>>indexed=model.transform(df)>>>indexed.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|+---+--------+-------------+8.3.2特征轉(zhuǎn)換：標(biāo)簽和索引的轉(zhuǎn)化IndexToString與StringIndexer相對(duì)應(yīng)，IndexToString的作用是把標(biāo)簽索引的一列重新映射回原有的字符型標(biāo)簽其主要使用場(chǎng)景一般都是和StringIndexer配合，先用StringIndexer將標(biāo)簽轉(zhuǎn)化成標(biāo)簽索引，進(jìn)行模型訓(xùn)練，然后在預(yù)測(cè)標(biāo)簽的時(shí)候再把標(biāo)簽索引轉(zhuǎn)化成原有的字符標(biāo)簽8.3.2特征轉(zhuǎn)換：標(biāo)簽和索引的轉(zhuǎn)化>>>frompyspark.ml.featureimportIndexToString,StringIndexer>>>toString=IndexToString(inputCol="categoryIndex",outputCol="originalCategory")>>>indexString=toString.transform(indexed)>>>indexString.select("id","originalCategory").show()+---+----------------+|id|originalCategory|+---+----------------+|0|a||1|b||2|c||3|a||4|a||5|c|+---+----------------+8.3.2特征轉(zhuǎn)換：標(biāo)簽和索引的轉(zhuǎn)化VectorIndexer之前介紹的StringIndexer是針對(duì)單個(gè)類(lèi)別型特征進(jìn)行轉(zhuǎn)換，倘若所有特征都已經(jīng)被組織在一個(gè)向量中，又想對(duì)其中某些單個(gè)分量進(jìn)行處理時(shí)，SparkML提供了VectorIndexer類(lèi)來(lái)解決向量數(shù)據(jù)集中的類(lèi)別性特征轉(zhuǎn)換通過(guò)為其提供maxCategories超參數(shù)，它可以自動(dòng)識(shí)別哪些特征是類(lèi)別型的，并且將原始值轉(zhuǎn)換為類(lèi)別索引。它基于不同特征值的數(shù)量來(lái)識(shí)別哪些特征需要被類(lèi)別化，那些取值可能性最多不超過(guò)maxCategories的特征需要會(huì)被認(rèn)為是類(lèi)別型的8.3.2特征轉(zhuǎn)換：標(biāo)簽和索引的轉(zhuǎn)化首先引入所需要的類(lèi)，并構(gòu)建數(shù)據(jù)集。>>>frompyspark.ml.featureimportVectorIndexer>>>frompyspark.ml.linalgimportVector,Vectors>>>df=spark.createDataFrame([\...(Vectors.dense(-1.0,1.0,1.0),),\...(Vectors.dense(-1.0,3.0,1.0),),\...(Vectors.dense(0.0,5.0,1.0),)],["features"])8.3.2特征轉(zhuǎn)換：標(biāo)簽和索引的轉(zhuǎn)化>>>indexer=VectorIndexer(inputCol="features",outputCol="indexed",maxCategories=2)>>>indexerModel=indexer.fit(df)然后，構(gòu)建VectorIndexer轉(zhuǎn)換器，設(shè)置輸入和輸出列，并進(jìn)行模型訓(xùn)練。8.3.2特征轉(zhuǎn)換：標(biāo)簽和索引的轉(zhuǎn)化>>>categoricalFeatures=indexerModel.categoryMaps.keys()>>>print("Choose"+str(len(categoricalFeatures))+\..."categoricalfeatures:"+str(categoricalFeatures))Chose2categoricalfeatures:[0,2]接下來(lái)，通過(guò)VectorIndexerModel的categoryMaps成員來(lái)獲得被轉(zhuǎn)換的特征及其映射，這里可以看到，共有兩個(gè)特征被轉(zhuǎn)換，分別是0號(hào)和2號(hào)。8.3.2特征轉(zhuǎn)換：標(biāo)簽和索引的轉(zhuǎn)化最后，把模型應(yīng)用于原有的數(shù)據(jù)，并打印結(jié)果。>>>indexed=indexerModel.transform(df)>>>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]|+--------------+-------------+8.4分類(lèi)與回歸8.4.1邏輯斯蒂回歸分類(lèi)器8.4.2決策樹(shù)分類(lèi)器8.4.1邏輯斯蒂回歸分類(lèi)器邏輯斯蒂回歸（logisticregression）是統(tǒng)計(jì)學(xué)習(xí)中的經(jīng)典分類(lèi)方法，屬于對(duì)數(shù)線性模型。logistic回歸的因變量可以是二分類(lèi)的，也可以是多分類(lèi)的。8.4.1邏輯斯蒂回歸分類(lèi)器任務(wù)描述：以iris數(shù)據(jù)集（iris）為例進(jìn)行分析（iris下載地址：/blog/wp-content/uploads/2017/03/iris.txt）iris以鳶尾花的特征作為數(shù)據(jù)來(lái)源，數(shù)據(jù)集包含150個(gè)數(shù)據(jù)集，分為3類(lèi)，每類(lèi)50個(gè)數(shù)據(jù)，每個(gè)數(shù)據(jù)包含4個(gè)屬性，是在數(shù)據(jù)挖掘、數(shù)據(jù)分類(lèi)中非常常用的測(cè)試集、訓(xùn)練集。為了便于理解，這里主要用后兩個(gè)屬性（花瓣的長(zhǎng)度和寬度）來(lái)進(jìn)行分類(lèi)。8.4.1邏輯斯蒂回歸分類(lèi)器首先我們先取其中的后兩類(lèi)數(shù)據(jù)，用二項(xiàng)邏輯斯蒂回歸進(jìn)行二分類(lèi)分析第1步：導(dǎo)入本地向量Vector和Vectors，導(dǎo)入所需要的類(lèi)。>>>frompyspark.ml.linalgimportVector,Vectors>>>frompyspark.sqlimportRow,functions>>>frompyspark.ml.evaluationimportMulticlassClassificationEvaluator>>>frompyspark.mlimportPipeline>>>frompyspark.ml.featureimportIndexToString,StringIndexer,\...VectorIndexer,HashingTF,Tokenizer>>>frompyspark.ml.classificationimportLogisticRegression,\...LogisticRegressionModel,BinaryLogisticRegressionSummary,LogisticRegression8.4.1邏輯斯蒂回歸分類(lèi)器2.第2步：我們定制一個(gè)函數(shù)，來(lái)返回一個(gè)指定的數(shù)據(jù)，然后讀取文本文件，第一個(gè)map把每行的數(shù)據(jù)用“,”隔開(kāi)，比如在我們的數(shù)據(jù)集中，每行被分成了5部分，前4部分是鳶尾花的4個(gè)特征，最后一部分是鳶尾花的分類(lèi)；我們這里把特征存儲(chǔ)在Vector中，創(chuàng)建一個(gè)Iris模式的RDD，然后轉(zhuǎn)化成dataframe；最后調(diào)用show()方法來(lái)查看一下部分?jǐn)?shù)據(jù)。>>>deff(x):...rel={}...rel['features']=Vectors.\...dense(float(x[0]),float(x[1]),float(x[2]),float(x[3]))...rel['label']=str(x[4])...returnrel剩余代碼見(jiàn)下一頁(yè)8.4.1邏輯斯蒂回歸分類(lèi)器>>>data=spark.sparkContext.\...textFile("file:///usr/local/spark/iris.txt").\...map(lambdaline:line.split(',')).\...map(lambdap:Row(**f(p))).\...toDF()>>>data.show()+-----------------+-----------+|features|label|+-----------------+-----------+|[5.1,3.5,1.4,0.2]|Iris-setosa||[4.9,3.0,1.4,0.2]|Iris-setosa||[4.7,3.2,1.3,0.2]|Iris-setosa||[4.6,3.1,1.5,0.2]|Iris-setosa|………+-----------------+-----------+onlyshowingtop20rows8.4.1邏輯斯蒂回歸分類(lèi)器3.第3步：分別獲取標(biāo)簽列和特征列，進(jìn)行索引并進(jìn)行重命名。>>>labelIndexer=StringIndexer().\...setInputCol("label").\...setOutputCol("indexedLabel").\...fit(data)>>>featureIndexer=VectorIndexer().\...setInputCol("features").\...setOutputCol("indexedFeatures").\...fit(data)8.4.1邏輯斯蒂回歸分類(lèi)器

第4步：設(shè)置LogisticRegression算法的參數(shù)。這里設(shè)置了循環(huán)次數(shù)為100次，規(guī)范化項(xiàng)為0.3等，具體可以設(shè)置的參數(shù)，可以通過(guò)explainParams()來(lái)獲取，還能看到程序已經(jīng)設(shè)置的參數(shù)的結(jié)果。>>>lr=LogisticRegression().\...setLabelCol("indexedLabel").\...setFeaturesCol("indexedFeatures").\...setMaxIter(100).\...setRegParam(0.3).\...setElasticNetParam(0.8)>>>print("LogisticRegressionparameters:\n"+lr.explainParams())8.4.1邏輯斯蒂回歸分類(lèi)器第5步：設(shè)置一個(gè)IndexToString的轉(zhuǎn)換器，把預(yù)測(cè)的類(lèi)別重新轉(zhuǎn)化成字符型的。構(gòu)建一個(gè)機(jī)器學(xué)習(xí)流水線，設(shè)置各個(gè)階段。上一個(gè)階段的輸出將是本階段的輸入。>>>labelConverter=IndexToString().\...setInputCol("prediction").\...setOutputCol("predictedLabel").\...setLabels(labelIndexer.labels)>>>lrPipeline=Pipeline().\...setStages([labelIndexer,featureIndexer,lr,labelConverter])8.4.1邏輯斯蒂回歸分類(lèi)器第6步：把數(shù)據(jù)集隨機(jī)分成訓(xùn)練集和測(cè)試集，其中訓(xùn)練集占70%。Pipeline本質(zhì)上是一個(gè)評(píng)估器，當(dāng)Pipeline調(diào)用fit()的時(shí)候就產(chǎn)生了一個(gè)PipelineModel，它是一個(gè)轉(zhuǎn)換器。然后，這個(gè)PipelineModel就可以調(diào)用transform()來(lái)進(jìn)行預(yù)測(cè)，生成一個(gè)新的DataFrame，即利用訓(xùn)練得到的模型對(duì)測(cè)試集進(jìn)行驗(yàn)證。>>>trainingData,testData=data.randomSplit([0.7,0.3])>>>lrPipelineModel=lrPipeline.fit(trainingData)>>>lrPredictions=lrPipelineModel.transform(testData)8.4.1邏輯斯蒂回歸分類(lèi)器第7步：輸出預(yù)測(cè)的結(jié)果，其中，select選擇要輸出的列，collect獲取所有行的數(shù)據(jù)，用foreach把每行打印出來(lái)。>>>preRel=lrPredictions.select(\..."predictedLabel",\..."label",\..."features",\..."probability").\...collect()>>>foriteminpreRel:...print(str(item['label'])+','+\...str(item['features'])+'-->prob='+\...str(item['probability'])+',predictedLabel'+\...str(item['predictedLabel']))

8.4.1邏輯斯蒂回歸分類(lèi)器第8步：對(duì)訓(xùn)練的模型進(jìn)行評(píng)估。創(chuàng)建一個(gè)MulticlassClassificationEvaluator實(shí)例，用setter方法把預(yù)測(cè)分類(lèi)的列名和真實(shí)分類(lèi)的列名進(jìn)行設(shè)置，然后計(jì)算預(yù)測(cè)準(zhǔn)確率。>>>evaluator=MulticlassClassificationEvaluator().\...setLabelCol("indexedLabel").\...setPredictionCol("prediction")>>>lrAccuracy=evaluator.evaluate(lrPredictions)>>>lrAccuracy0.7774712643678161#模型預(yù)測(cè)的準(zhǔn)確率8.4.1邏輯斯蒂回歸分類(lèi)器第9步：可以通過(guò)model來(lái)獲取訓(xùn)練得到的邏輯斯蒂模型。lrPipelineModel是一個(gè)PipelineModel，因此，可以通過(guò)調(diào)用它的stages方法來(lái)獲取模型，具體如下：>>>lrModel=lrPipelineModel.stages[2]>>>print("Coefficients:\n"+str(lrModel.coefficientMatrix)+\..."\nIntercept:"+str(lrMerceptVector)+\..."\nnumClasses:"+str(lrModel.numClasses)+\..."\nnumFeatures:"+str(lrModel.numFeatures))

Coefficients:3X4CSRMatrix(1,3)0.4332(2,2)-0.2472(2,3)-0.1689Intercept:[-0.11530503231364186,-0.63496556499483,0.750270597308472]numClasses:3numFeatures:48.4.2決策樹(shù)分類(lèi)器決策樹(shù)（decisiontree）是一種基本的分類(lèi)與回歸方法，這里主要介紹用于分類(lèi)的決策樹(shù)。決策樹(shù)模式呈樹(shù)形結(jié)構(gòu)，其中每個(gè)內(nèi)部節(jié)點(diǎn)表示一個(gè)屬性上的測(cè)試，每個(gè)分支代表一個(gè)測(cè)試輸出，每個(gè)葉節(jié)點(diǎn)代表一種類(lèi)別。學(xué)習(xí)時(shí)利用訓(xùn)練數(shù)據(jù)，根據(jù)損失函數(shù)最小化的原則建立決策樹(shù)模型；預(yù)測(cè)時(shí)，對(duì)新的數(shù)據(jù)，利用決策樹(shù)模型進(jìn)行分類(lèi)決策樹(shù)學(xué)習(xí)通常包括3個(gè)步驟：特征選擇、決策樹(shù)的生成和決策樹(shù)的剪枝8.4.2決策樹(shù)分類(lèi)器我們以iris數(shù)據(jù)集（iris）為例進(jìn)行分析（iris下載地址：/blog/wp-content/uploads/2017/03/iris.txt）iris以鳶尾花的特征作為數(shù)據(jù)來(lái)源，數(shù)據(jù)集包含150個(gè)數(shù)據(jù)集，分為3類(lèi)，每類(lèi)50個(gè)數(shù)據(jù)，每個(gè)數(shù)據(jù)包含4個(gè)屬性，是在數(shù)據(jù)挖掘、數(shù)據(jù)分類(lèi)中非常常用的測(cè)試集、訓(xùn)練集。8.4.2決策樹(shù)分類(lèi)器1.導(dǎo)入需要的包>>>frompyspark.ml.classificationimportDecisionTreeClassificationModel>>>frompyspark.ml.classificationimportDecisionTreeClassifier>>>frompyspark.mlimportPipeline,PipelineModel>>>frompyspark.ml.evaluationimportMulticlassClassificationEvaluator>>>frompyspark.ml.linalgimportVector,Vectors>>>frompyspark.sqlimportRow>>>frompyspark.ml.featureimportIndexToString,StringIndexer,VectorIndexer8.4.2決策樹(shù)分類(lèi)器2.第2步：讀取文本文件，第一個(gè)map把每行的數(shù)據(jù)用“,”隔開(kāi)，比如在我們的數(shù)據(jù)集中，每行被分成了5部分，前4部分是鳶尾花的4個(gè)特征，最后一部分是鳶尾花的分類(lèi)；我們這里把特征存儲(chǔ)在Vector中，創(chuàng)建一個(gè)Iris模式的RDD，然后轉(zhuǎn)化成dataframe。>>>deff(x):...rel={}...rel['features']=Vectors.\...dense(float(x[0]),float(x[1]),float(x[2]),float(x[3]))...rel['label']=str(x[4])...returnrel>>>data=spark.sparkContext.\...textFile("file:///usr/local/spark/iris.txt").\...map(lambdaline:line.split(',')).\...map(lambdap:Row(**f(p))).\...toDF()8.4.2決策樹(shù)分類(lèi)器>>>labelIndexer=StringIndexer().\...setInputCol("label").\...setOutputCol("indexedLabel").\...fit(data)>>>featureIndexer=VectorIndexer().\...setInputCol("features").\...setOutputCol("indexedFeatures").\...setMaxCategories(4).\...fit(data)>>>labelConverter=IndexToString().\...setInputCol("prediction").\