廈門大學計算機科學系2019版

第8章SparkMLlib

（PPT版本號：2019年春季學期）

溫馨提示：編輯幻燈片母版，可以修改每頁PPT的廈大校徽和底部文字《Spark編程基礎（Python版）》教材官網：/post/spark-python/掃一掃訪問教材官網林子雨廈門大學計算機科學系E-mail:ziyulin@主頁：/linziyu課程教材Spark入門教程(Python版)/blog/1709-2/掃一掃訪問在線教程紙質教材預期在2019年夏天上市銷售本書以Python作為開發Spark應用程序的編程語言，系統介紹了Spark編程的基礎知識。全書共8章，內容包括大數據技術概述、Spark的設計與運行原理、Spark環境搭建和使用方法、RDD編程、SparkSQL、SparkStreaming、StructuredStreaming、SparkMLlib等。本書每個章節都安排了入門級的編程實踐操作，以便讀者更好地學習和掌握Spark編程方法。本書官網免費提供了全套的在線教學資源，包括講義PPT、習題、源代碼、軟件、數據集、授課視頻、上機實驗指南等。提綱8.1SparkMLlib簡介8.2機器學習工作流8.3特征抽取、轉化和選擇8.4分類與回歸百度搜索廈門大學數據庫實驗室網站訪問平臺8.1SparkMLlib簡介8.1.1什么是機器學習8.1.2基于大數據的機器學習8.1.3Spark機器學習庫MLLib8.1.1什么是機器學習機器學習可以看做是一門人工智能的科學，該領域的主要研究對象是人工智能。機器學習利用數據或以往的經驗，以此優化計算機程序的性能標準。機器學習強調三個關鍵詞：算法、經驗、性能8.1.2基于大數據的機器學習傳統的機器學習算法，由于技術和單機存儲的限制，只能在少量數據上使用，依賴于數據抽樣大數據技術的出現，可以支持在全量數據上進行機器學習機器學習算法涉及大量迭代計算基于磁盤的MapReduce不適合進行大量迭代計算基于內存的Spark比較適合進行大量迭代計算8.1.3Spark機器學習庫MLlibSpark提供了一個基于海量數據的機器學習庫，它提供了常用機器學習算法的分布式實現開發者只需要有Spark基礎并且了解機器學習算法的原理，以及方法相關參數的含義，就可以輕松的通過調用相應的API來實現基于海量數據的機器學習過程pyspark的即席查詢也是一個關鍵。算法工程師可以邊寫代碼邊運行，邊看結果8.1.3Spark機器學習庫MLlib需要注意的是，MLlib中只包含能夠在集群上運行良好的并行算法，這一點很重要有些經典的機器學習算法沒有包含在其中，就是因為它們不能并行執行相反地，一些較新的研究得出的算法因為適用于集群，也被包含在MLlib中，例如分布式隨機森林算法、最小交替二乘算法。這樣的選擇使得MLlib中的每一個算法都適用于大規模數據集如果是小規模數據集上訓練各機器學習模型，最好還是在各個節點上使用單節點的機器學習算法庫（比如Weka）8.1.3Spark機器學習庫MLlibMLlib是Spark的機器學習（MachineLearning）庫，旨在簡化機器學習的工程實踐工作MLlib由一些通用的學習算法和工具組成，包括分類、回歸、聚類、協同過濾、降維等，同時還包括底層的優化原語和高層的流水線（Pipeline）API，具體如下：算法工具：常用的學習算法，如分類、回歸、聚類和協同過濾；特征化工具：特征提取、轉化、降維和選擇工具；流水線(Pipeline)：用于構建、評估和調整機器學習工作流的工具;持久性：保存和加載算法、模型和管道;實用工具：線性代數、統計、數據處理等工具。8.1.3Spark機器學習庫MLlibSpark機器學習庫從1.2版本以后被分為兩個包：spark.mllib

包含基于RDD的原始算法API。SparkMLlib歷史比較長，在1.0以前的版本即已經包含了，提供的算法實現都是基于原始的RDDspark.ml則提供了基于DataFrames高層次的API，可以用來構建機器學習工作流（PipeLine）。MLPipeline彌補了原始MLlib庫的不足，向用戶提供了一個基于DataFrame的機器學習工作流式API套件8.1.3Spark機器學習庫MLlibMLlib目前支持4種常見的機器學習問題:

分類、回歸、聚類和協同過濾8.2機器學習流水線8.2.1機器學習流水線概念8.2.2構建一個機器學習流水線8.2.1機器學習流水線概念在介紹流水線之前，先來了解幾個重要概念：DataFrame：使用SparkSQL中的DataFrame作為數據集，它可以容納各種數據類型。較之RDD，DataFrame包含了schema信息，更類似傳統數據庫中的二維表格。它被MLPipeline用來存儲源數據。例如，DataFrame中的列可以是存儲的文本、特征向量、真實標簽和預測的標簽等8.2.1機器學習流水線概念Transformer：翻譯成轉換器，是一種可以將一個DataFrame轉換為另一個DataFrame的算法。比如一個模型就是一個Transformer。它可以把一個不包含預測標簽的測試數據集DataFrame打上標簽，轉化成另一個包含預測標簽的DataFrame。技術上，Transformer實現了一個方法transform()，它通過附加一個或多個列將一個DataFrame轉換為另一個DataFrame8.2.1機器學習流水線概念Estimator：翻譯成估計器或評估器，它是學習算法或在訓練數據上的訓練方法的概念抽象。在Pipeline里通常是被用來操作DataFrame數據并生成一個Transformer。從技術上講，Estimator實現了一個方法fit()，它接受一個DataFrame并產生一個轉換器。比如，一個隨機森林算法就是一個Estimator，它可以調用fit()，通過訓練特征數據而得到一個隨機森林模型。8.2.1機器學習流水線概念Parameter：Parameter被用來設置Transformer或者Estimator的參數。現在，所有轉換器和估計器可共享用于指定參數的公共API。ParamMap是一組（參數，值）對PipeLine：翻譯為流水線或者管道。流水線將多個工作流階段（轉換器和估計器）連接在一起，形成機器學習的工作流，并獲得結果輸出8.2.2流水線工作過程要構建一個Pipeline流水線，首先需要定義Pipeline中的各個流水線階段PipelineStage（包括轉換器和評估器），比如指標提取和轉換模型訓練等。有了這些處理特定問題的轉換器和評估器，就可以按照具體的處理邏輯有序地組織PipelineStages并創建一個Pipeline>>>pipeline=Pipeline(stages=[stage1,stage2,stage3])然后就可以把訓練數據集作為輸入參數，調用Pipeline實例的fit方法來開始以流的方式來處理源訓練數據。這個調用會返回一個PipelineModel類實例，進而被用來預測測試數據的標簽8.2.2流水線工作過程流水線的各個階段按順序運行，輸入的DataFrame在它通過每個階段時被轉換8.2.2流水線工作過程值得注意的是，流水線本身也可以看做是一個估計器。在流水線的fit（）方法運行之后，它產生一個PipelineModel，它是一個Transformer。這個管道模型將在測試數據的時候使用。下圖說明了這種用法。8.2.3構建一個機器學習流水線任務描述查找出所有包含"spark"的句子，即將包含"spark"的句子的標簽設為1，沒有"spark"的句子的標簽設為0。本節以邏輯斯蒂回歸為例，構建一個典型的機器學習過程，來具體介紹一下流水線是如何應用的8.2.3構建一個機器學習流水線需要使用SparkSession對象Spark2.0以上版本的pyspark在啟動時會自動創建一個名為spark的SparkSession對象，當需要手工創建時，SparkSession可以由其伴生對象的builder()方法創建出來，如下代碼段所示：frompyspark.sqlimportSparkSessionspark=SparkSession.builder.master("local").appName("WordCount").getOrCreate()pyspark.ml依賴numpy包，Ubuntu自帶python3是沒有numpy的，執行如下命令安裝：sudopip3installnumpy8.2.3構建一個機器學習流水線（1）引入要包含的包并構建訓練數據集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構建一個機器學習流水線（2）定義Pipeline中的各個流水線階段PipelineStage，包括轉換器和評估器，具體地，包含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構建一個機器學習流水線（3）按照具體的處理邏輯有序地組織PipelineStages，并創建一個Pipeline。pipeline=Pipeline(stages=[tokenizer,hashingTF,lr])現在構建的Pipeline本質上是一個Estimator，在它的fit()方法運行之后，它將產生一個PipelineModel，它是一個Transformer。model=pipeline.fit(training)可以看到，model的類型是一個PipelineModel，這個流水線模型將在測試數據的時候使用8.2.3構建一個機器學習流水線（4）構建測試數據test=spark.createDataFrame([(4,"sparkijk"),(5,"lmn"),(6,"sparkhadoopspark"),(7,"apachehadoop")],["id","text"])8.2.3構建一個機器學習流水線（5）調用之前訓練好的PipelineModel的transform()方法，讓測試數據按順序通過擬合的流水線，生成預測結果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.130000000004,0.130000000006],prediction=1.000000(5,lmn)-->prob=[0.130000000001,0.130000000009],prediction=0.000000(6,sparkhadoopspark)-->prob=[0.1300000000095,0,prediction=1.000000(7,apachehadoop)-->prob=[0.130000000001,0.130000000009],prediction=0.0000008.3特征提取、轉換和選擇8.3.1特征提取8.3.2特征轉換8.3.3特征選取8.3.1特征抽取：TF-IDF“詞頻－逆向文件頻率”（TF-IDF）是一種在文本挖掘中廣泛使用的特征向量化方法，它可以體現一個文檔中詞語在語料庫中的重要程度。詞語由t表示，文檔由d表示，語料庫由D表示。詞頻TF(t,d)是詞語t在文檔d中出現的次數。文件頻率DF(t,D)是包含詞語的文檔的個數。TF-IDF就是在數值化文檔信息，衡量詞語能提供多少信息以區分文檔。其定義如下：TF-IDF度量值表示如下：8.3.1特征抽取：TF-IDF在SparkML庫中，TF-IDF被分成兩部分：TF(+hashing)

IDFTF:

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

IDF是一個Estimator，在一個數據集上應用它的fit()方法，產生一個IDFModel。該IDFModel接收特征向量（由HashingTF產生），然后計算每一個詞在文檔中出現的頻次。IDF會減少那些在語料庫中出現頻率較高的詞的權重。8.3.1特征抽取：TF-IDF過程描述：在下面的代碼段中，我們以一組句子開始首先使用分解器Tokenizer把句子劃分為單個詞語對每一個句子（詞袋），使用HashingTF將句子轉換為特征向量最后使用IDF重新調整特征向量（這種轉換通常可以提高使用文本特征的性能）8.3.1特征抽取：TF-IDF（1）導入TF-IDF所需要的包：>>>frompyspark.ml.featureimportHashingTF,IDF,Tokenizer（2）創建一個簡單的DataFrame，每一個句子代表一個文檔>>>sentenceData=spark.createDataFrame([(0,"IheardaboutSparkandIloveSpark"),(0,"IwishJavacouldusecaseclasses"),(1,"Logisticregressionmodelsareneat")]).toDF("label","sentence")8.3.1特征抽取：TF-IDF（3）得到文檔集合后，即可用tokenizer對句子進行分詞>>>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特征抽取：TF-IDF（4）得到分詞后的文檔序列后，即可使用HashingTF的transform()方法把句子哈希成特征向量，這里設置哈希表的桶數為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特征抽取：TF-IDF（5）調用IDF方法來重新構造特征向量的規模，生成的變量idf是一個評估器，在特征向量上應用它的fit()方法，會產生一個IDFModel（名稱為idfModel）。>>>idf=IDF(inputCol="rawFeatures",outputCol="features")>>>idfModel=idf.fit(featurizedData)8.3.1特征抽取：TF-IDF>>>rescaledData=idfModel.transform(featurizedData)>>>rescaledData.select("features","label").show(truncate=False)+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-----+|features|label|+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-----+|(2000,[240,333,1105,1329,1357,1777],[0.1300000000099453,0.1300000000099453,1.1300000000098906,0.1300000000035617,0.1300000000099453,0.1300000000099453])|0||(2000,[213,342,489,495,1329,1809,1967],[0.1300000000099453,0.1300000000099453,0.1300000000099453,0.1300000000099453,0.13000000000178085,0.1300000000099453,0.1300000000099453])|0||(2000,[286,695,1138,1193,1604],[0.1300000000099453,0.1300000000099453,0.1300000000099453,0.1300000000099453,0.1300000000099453])|1|+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-----+（6）調用IDFModel的transform()方法，可以得到每一個單詞對應的TF-IDF度量值。8.3.2特征轉換：標簽和索引的轉化在機器學習處理過程中，為了方便相關算法的實現，經常需要把標簽數據（一般是字符串）轉化成整數索引，或是在計算結束后將整數索引還原為相應的標簽SparkML包中提供了幾個相關的轉換器，例如：StringIndexer、IndexToString、OneHotEncoder、VectorIndexer，它們提供了十分方便的特征轉換功能，這些轉換器類都位于org.apache.spark.ml.feature包下值得注意的是，用于特征轉換的轉換器和其他的機器學習算法一樣，也屬于MLPipeline模型的一部分，可以用來構成機器學習流水線，以StringIndexer為例，其存儲著進行標簽數值化過程的相關超參數，是一個Estimator，對其調用fit(..)方法即可生成相應的模型StringIndexerModel類，很顯然，它存儲了用于DataFrame進行相關處理的參數，是一個Transformer（其他轉換器也是同一原理）8.3.2特征轉換：標簽和索引的轉化StringIndexerStringIndexer轉換器可以把一列類別型的特征（或標簽）進行編碼，使其數值化，索引的范圍從0開始，該過程可以使得相應的特征索引化，使得某些無法接受類別型特征的算法可以使用，并提高諸如決策樹等機器學習算法的效率索引構建的順序為標簽的頻率，優先編碼頻率較大的標簽，所以出現頻率最高的標簽為0號如果輸入的是數值型的，會首先把它轉化成字符型，然后再對其進行編碼8.3.2特征轉換：標簽和索引的轉化（1）首先，引入所需要使用的類。>>>frompyspark.ml.featureimportStringIndexer8.3.2特征轉換：標簽和索引的轉化（2）其次，構建1個DataFrame，設置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特征轉換：標簽和索引的轉化（3）然后，通過fit()方法進行模型訓練，用訓練出的模型對原數據集進行處理，并通過indexed.show()進行展示。>>>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特征轉換：標簽和索引的轉化IndexToString與StringIndexer相對應，IndexToString的作用是把標簽索引的一列重新映射回原有的字符型標簽其主要使用場景一般都是和StringIndexer配合，先用StringIndexer將標簽轉化成標簽索引，進行模型訓練，然后在預測標簽的時候再把標簽索引轉化成原有的字符標簽8.3.2特征轉換：標簽和索引的轉化>>>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特征轉換：標簽和索引的轉化VectorIndexer之前介紹的StringIndexer是針對單個類別型特征進行轉換，倘若所有特征都已經被組織在一個向量中，又想對其中某些單個分量進行處理時，SparkML提供了VectorIndexer類來解決向量數據集中的類別性特征轉換通過為其提供maxCategories超參數，它可以自動識別哪些特征是類別型的，并且將原始值轉換為類別索引。它基于不同特征值的數量來識別哪些特征需要被類別化，那些取值可能性最多不超過maxCategories的特征需要會被認為是類別型的8.3.2特征轉換：標簽和索引的轉化首先引入所需要的類，并構建數據集。>>>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特征轉換：標簽和索引的轉化>>>indexer=VectorIndexer(inputCol="features",outputCol="indexed",maxCategories=2)>>>indexerModel=indexer.fit(df)然后，構建VectorIndexer轉換器，設置輸入和輸出列，并進行模型訓練。8.3.2特征轉換：標簽和索引的轉化>>>categoricalFeatures=indexerModel.categoryMaps.keys()>>>print("Choose"+str(len(categoricalFeatures))+\..."categoricalfeatures:"+str(categoricalFeatures))Chose2categoricalfeatures:[0,2]接下來，通過VectorIndexerModel的categoryMaps成員來獲得被轉換的特征及其映射，這里可以看到，共有兩個特征被轉換，分別是0號和2號。8.3.2特征轉換：標簽和索引的轉化最后，把模型應用于原有的數據，并打印結果。>>>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分類與回歸8.4.1邏輯斯蒂回歸分類器8.4.2決策樹分類器8.4.1邏輯斯蒂回歸分類器邏輯斯蒂回歸（logisticregression）是統計學習中的經典分類方法，屬于對數線性模型。logistic回歸的因變量可以是二分類的，也可以是多分類的。8.4.1邏輯斯蒂回歸分類器任務描述：以iris數據集（iris）為例進行分析（iris下載地址：/blog/wp-content/uploads/2017/03/iris.txt）iris以鳶尾花的特征作為數據來源，數據集包含150個數據集，分為3類，每類50個數據，每個數據包含4個屬性，是在數據挖掘、數據分類中非常常用的測試集、訓練集。為了便于理解，這里主要用后兩個屬性（花瓣的長度和寬度）來進行分類。8.4.1邏輯斯蒂回歸分類器首先我們先取其中的后兩類數據，用二項邏輯斯蒂回歸進行二分類分析第1步：導入本地向量Vector和Vectors，導入所需要的類。>>>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邏輯斯蒂回歸分類器2.第2步：我們定制一個函數，來返回一個指定的數據，然后讀取文本文件，第一個map把每行的數據用“,”隔開，比如在我們的數據集中，每行被分成了5部分，前4部分是鳶尾花的4個特征，最后一部分是鳶尾花的分類；我們這里把特征存儲在Vector中，創建一個Iris模式的RDD，然后轉化成dataframe；最后調用show()方法來查看一下部分數據。>>>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剩余代碼見下一頁8.4.1邏輯斯蒂回歸分類器>>>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邏輯斯蒂回歸分類器3.第3步：分別獲取標簽列和特征列，進行索引并進行重命名。>>>labelIndexer=StringIndexer().\...setInputCol("label").\...setOutputCol("indexedLabel").\...fit(data)>>>featureIndexer=VectorIndexer().\...setInputCol("features").\...setOutputCol("indexedFeatures").\...fit(data)8.4.1邏輯斯蒂回歸分類器

第4步：設置LogisticRegression算法的參數。這里設置了循環次數為100次，規范化項為0.3等，具體可以設置的參數，可以通過explainParams()來獲取，還能看到程序已經設置的參數的結果。>>>lr=LogisticRegression().\...setLabelCol("indexedLabel").\...setFeaturesCol("indexedFeatures").\...setMaxIter(100).\...setRegParam(0.3).\...setElasticNetParam(0.8)>>>print("LogisticRegressionparameters:\n"+lr.explainParams())8.4.1邏輯斯蒂回歸分類器第5步：設置一個IndexToString的轉換器，把預測的類別重新轉化成字符型的。構建一個機器學習流水線，設置各個階段。上一個階段的輸出將是本階段的輸入。>>>labelConverter=IndexToString().\...setInputCol("prediction").\...setOutputCol("predictedLabel").\...setLabels(labelIndexer.labels)>>>lrPipeline=Pipeline().\...setStages([labelIndexer,featureIndexer,lr,labelConverter])8.4.1邏輯斯蒂回歸分類器第6步：把數據集隨機分成訓練集和測試集，其中訓練集占70%。Pipeline本質上是一個評估器，當Pipeline調用fit()的時候就產生了一個PipelineModel，它是一個轉換器。然后，這個PipelineModel就可以調用transform()來進行預測，生成一個新的DataFrame，即利用訓練得到的模型對測試集進行驗證。>>>trainingData,testData=data.randomSplit([0.7,0.3])>>>lrPipelineModel=lrPipeline.fit(trainingData)>>>lrPredictions=lrPipelineModel.transform(testData)8.4.1邏輯斯蒂回歸分類器第7步：輸出預測的結果，其中，select選擇要輸出的列，collect獲取所有行的數據，用foreach把每行打印出來。>>>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邏輯斯蒂回歸分類器第8步：對訓練的模型進行評估。創建一個MulticlassClassificationEvaluator實例，用setter方法把預測分類的列名和真實分類的列名進行設置，然后計算預測準確率。>>>evaluator=MulticlassClassificationEvaluator().\...setLabelCol("indexedLabel").\...setPredictionCol("prediction")>>>lrAccuracy=evaluator.evaluate(lrPredictions)>>>lrAccuracy0.1300000000078161#模型預測的準確率8.4.1邏輯斯蒂回歸分類器第9步：可以通過model來獲取訓練得到的邏輯斯蒂模型。lrPipelineModel是一個PipelineModel，因此，可以通過調用它的stages方法來獲取模型，具體如下：>>>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.13000000000364186,-0.13000000000483,0.130000000008472]numClasses:3numFeatures:48.4.2決策樹分類器決策樹（decisiontree）是一種基本的分類與回歸方法，這里主要介紹用于分類的決策樹。決策樹模式呈樹形結構，其中每個內部節點表示一個屬性上的測試，每個分支代表一個測試輸出，每個葉節點代表一種類別。學習時利用訓練數據，根據損失函數最小化的原則建立決策樹模型；預測時，對新的數據，利用決策樹模型進行分類決策樹學習通常包括3個步驟：特征選擇、決策樹的生成和決策樹的剪枝8.4.2決策樹分類器我們以iris數據集（iris）為例進行分析（iris下載地址：/blog/wp-content/uploads/2017/03/iris.txt）iris以鳶尾花的特征作為數據來源，數據集包含150個數據集，分為3類，每類50個數據，每個數據包含4個屬性，是在數據挖掘、數據分類中非常常用的測試集、訓練集。8.4.2決策樹分類器1.導入需要的包>>>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決策樹分類器2.第2步：讀取文本文件，第一個map把每行的數據用“,”隔開，比如在我們的數據集中，每行被分成了5部分，前4部分是鳶尾花的4個特征，最后一部分是鳶尾花的分類；我們這里把特征存儲在Vector中，創建一個Iris模式的RDD，然后轉化成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決策樹分類器>>>labelIndexer=StringIndexer().\...setInputCol("label").\...setOutputCol("indexedLabel").\...fit(data)>>>featureIndexer=VectorIndexer().\...setInputCol("features").\...setOutputCol("indexedFeatures").\...setMaxCategories(4).\...fit(data)>>>labelConverter=IndexToString().\...setInputCol("prediction").\...setOutputCol("predictedLabel").\...setLabels(labelIndexer.labels)>>>trainingData,testData=data.randomSplit([0.7,0.3])第3步：進一步處理特征和標簽，把數據集隨機分成訓練集和測試集，其中訓練集占70%。8.4.2決策樹分類器第4步：創建決策樹模型DecisionTreeClassifier，通過setter的方法來設置決策樹的參數，也可以用ParamMap來設置。這里僅需要設置特征列（FeaturesCol）和待預測列（LabelCol）。具體可以設置的參數可以通過explainParams()來獲取。>>>dtClassifier=DecisionTreeClassifier().\...setLabelCol("indexedLabel").\...setFeaturesCol("indexedFeatures")8.4.2決策樹分類器>>>dtPipeline=Pipeline().\...setStages([labelIndexer,featureIndexer,dtClassifier,labelConverter])>>>dtPipelineModel=dtPipeline.fit(trainingData)>>>dtPredictions=dtPipelineModel.transform(testData)>>>dtPredictions.select("predictedLabel","label","features").show(20)+---------------+---------------+-----------------+|predictedLabel|label|features|+---------------+---------------+-----------------+|Iris-setosa|Iris-setosa|[4.4,3.0,1.3,0.2]||Iris-setosa|Iris-setosa|[4.6,3.4,1.4,0.3]||Iris-setosa|Iris-setosa|[4.9,3.1,1.5,0.1]||Iris-setosa|Iris-setosa|[5.0,3.2,1.2,0.2]|第5步：構建機器學習流水線（Pipeline），在訓練數據集上調用fit()進行模型訓練，并在測試數據集上調用transform()方法進行預測。剩余代碼見下一頁8.4.2決策樹分類器>>>evaluator=MulticlassClassificationEvaluator().\...setLabelCol("indexedLabel").\...setPredictionCol("prediction")>>>dtAccuracy=evaluator.evaluate(dtPredictions)>>>dtAccuracy0.1300000000003888#模型的預測準確率8.4.2決策樹分類器>>>treeModelClassifier=dtPipelineModel.stages[2]>>>print("Learnedclassificationtreemodel:\n"+\...str(treeModelClassifier.toDebugString))

Learnedclassificationtreemodel:DecisionTreeClassificationModel(uid=DecisionTreeClassifier_5427198bb4c1)ofdepth5with15nodesIf(feature2<=2.45)Predict:2.0Else(feature2>2.45)If(feature2<=4.75)Predict:0.0Else(feature2>4.75)If(feature3<=1.75)If(feature2<=4.95)……第6步：可以通過調用DecisionTreeClassificationModel的toDebugString方法，查看訓練的決策樹模型結構。本章小結1、機器學習、SparkMLlib的基本概念2、機器學習工作流

構建一個機器學習工作流、特