Spark編程基礎Python版-第8章-Spark-MLlib

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

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

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

IDFTF:

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

IDF是一個Estimator，在一個數(shù)據(jù)集上應用它的fit()方法，產(chǎn)生一個IDFModel。該IDFModel接收特征向量（由HashingTF產(chǎn)生），然后計算每一個詞在文檔中出現(xiàn)的頻次。IDF會減少那些在語料庫中出現(xiàn)頻率較高的詞的權重。8.3.1特征提?。篢F-IDF過程描述：在下面的代碼段中，我們以一組句子開始首先使用分解器Tokenizer把句子劃分為單個詞語對每一個句子（詞袋），使用HashingTF將句子轉(zhuǎn)換為特征向量最后使用IDF重新調(diào)整特征向量（這種轉(zhuǎn)換通?？梢蕴岣呤褂梦谋咎卣鞯男阅埽?.3.1特征提?。篢F-IDF（1）導入TF-IDF所需要的包：>>>frompyspark.ml.featureimportHashingTF,IDF,Tokenizer（2）創(chuàng)建一個簡單的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()方法把句子哈希成特征向量，這里設置哈希表的桶數(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方法來重新構造特征向量的規(guī)模，生成的變量idf是一個評估器，在特征向量上應用它的fit()方法，會產(chǎn)生一個IDFModel（名稱為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()方法，可以得到每一個單詞對應的TF-IDF度量值。8.3.2特征轉(zhuǎn)換：標簽和索引的轉(zhuǎn)化在機器學習處理過程中，為了方便相關算法的實現(xiàn)，經(jīng)常需要把標簽數(shù)據(jù)（一般是字符串）轉(zhuǎn)化成整數(shù)索引，或是在計算結束后將整數(shù)索引還原為相應的標簽SparkML包中提供了幾個相關的轉(zhuǎn)換器，例如：StringIndexer、IndexToString、OneHotEncoder、VectorIndexer，它們提供了十分方便的特征轉(zhuǎn)換功能，這些轉(zhuǎn)換器類都位于org.apache.spark.ml.feature包下值得注意的是，用于特征轉(zhuǎn)換的轉(zhuǎn)換器和其他的機器學習算法一樣，也屬于MLPipeline模型的一部分，可以用來構成機器學習流水線，以StringIndexer為例，其存儲著進行標簽數(shù)值化過程的相關超參數(shù)，是一個Estimator，對其調(diào)用fit(..)方法即可生成相應的模型StringIndexerModel類，很顯然，它存儲了用于DataFrame進行相關處理的參數(shù)，是一個Transformer（其他轉(zhuǎn)換器也是同一原理）8.3.2特征轉(zhuǎn)換：標簽和索引的轉(zhuǎn)化StringIndexerStringIndexer轉(zhuǎn)換器可以把一列類別型的特征（或標簽）進行編碼，使其數(shù)值化，索引的范圍從0開始，該過程可以使得相應的特征索引化，使得某些無法接受類別型特征的算法可以使用，并提高諸如決策樹等機器學習算法的效率索引構建的順序為標簽的頻率，優(yōu)先編碼頻率較大的標簽，所以出現(xiàn)頻率最高的標簽為0號如果輸入的是數(shù)值型的，會首先把它轉(zhuǎn)化成字符型，然后再對其進行編碼8.3.2特征轉(zhuǎn)換：標簽和索引的轉(zhuǎn)化（1）首先，引入所需要使用的類。>>>frompyspark.ml.featureimportStringIndexer8.3.2特征轉(zhuǎn)換：標簽和索引的轉(zhuǎn)化（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特征轉(zhuǎn)換：標簽和索引的轉(zhuǎn)化（3）然后，通過fit()方法進行模型訓練，用訓練出的模型對原數(shù)據(jù)集進行處理，并通過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特征轉(zhuǎn)換：標簽和索引的轉(zhuǎn)化IndexToString與StringIndexer相對應，IndexToString的作用是把標簽索引的一列重新映射回原有的字符型標簽其主要使用場景一般都是和StringIndexer配合，先用StringIndexer將標簽轉(zhuǎn)化成標簽索引，進行模型訓練，然后在預測標簽的時候再把標簽索引轉(zhuǎn)化成原有的字符標簽8.3.2特征轉(zhuǎn)換：標簽和索引的轉(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)換：標簽和索引的轉(zhuǎn)化VectorIndexer之前介紹的StringIndexer是針對單個類別型特征進行轉(zhuǎn)換，倘若所有特征都已經(jīng)被組織在一個向量中，又想對其中某些單個分量進行處理時，SparkML提供了VectorIndexer類來解決向量數(shù)據(jù)集中的類別性特征轉(zhuǎn)換通過為其提供maxCategories超參數(shù)，它可以自動識別哪些特征是類別型的，并且將原始值轉(zhuǎn)換為類別索引。它基于不同特征值的數(shù)量來識別哪些特征需要被類別化，那些取值可能性最多不超過maxCategories的特征需要會被認為是類別型的8.3.2特征轉(zhuǎn)換：標簽和索引的轉(zhuǎn)化首先引入所需要的類，并構建數(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)換：標簽和索引的轉(zhuǎn)化>>>indexer=VectorIndexer(inputCol="features",outputCol="indexed",maxCategories=2)>>>indexerModel=indexer.fit(df)然后，構建VectorIndexer轉(zhuǎn)換器，設置輸入和輸出列，并進行模型訓練。8.3.2特征轉(zhuǎn)換：標簽和索引的轉(zhuǎn)化>>>categoricalFeatures=indexerModel.categoryMaps.keys()>>>print("Choose"+str(len(categoricalFeatures))+\..."categoricalfeatures:"+str(categoricalFeatures))Chose2categoricalfeatures:[0,2]接下來，通過VectorIndexerModel的categoryMaps成員來獲得被轉(zhuǎn)換的特征及其映射，這里可以看到，共有兩個特征被轉(zhuǎn)換，分別是0號和2號。8.3.2特征轉(zhuǎn)換：標簽和索引的轉(zhuǎn)化最后，把模型應用于原有的數(shù)據(jù)，并打印結果。>>>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）是統(tǒng)計學習中的經(jīng)典分類方法，屬于對數(shù)線性模型。logistic回歸的因變量可以是二分類的，也可以是多分類的。8.4.1邏輯斯蒂回歸分類器任務描述：以iris數(shù)據(jù)集（iris）為例進行分析（iris下載地址：/blog/wp-content/uploads/2017/03/iris.txt）iris以鳶尾花的特征作為數(shù)據(jù)來源，數(shù)據(jù)集包含150個數(shù)據(jù)集，分為3類，每類50個數(shù)據(jù)，每個數(shù)據(jù)包含4個屬性，是在數(shù)據(jù)挖掘、數(shù)據(jù)分類中非常常用的測試集、訓練集。為了便于理解，這里主要用后兩個屬性（花瓣的長度和寬度）來進行分類。8.4.1邏輯斯蒂回歸分類器首先我們先取其中的后兩類數(shù)據(jù)，用二項邏輯斯蒂回歸進行二分類分析第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步：我們定制一個函數(shù)，來返回一個指定的數(shù)據(jù)，然后讀取文本文件，第一個map把每行的數(shù)據(jù)用“,”隔開，比如在我們的數(shù)據(jù)集中，每行被分成了5部分，前4部分是鳶尾花的4個特征，最后一部分是鳶尾花的分類；我們這里把特征存儲在Vector中，創(chuàng)建一個Iris模式的RDD，然后轉(zhuǎn)化成dataframe；最后調(diào)用show()方法來查看一下部分數(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剩余代碼見下一頁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算法的參數(shù)。這里設置了循環(huán)次數(shù)為100次，規(guī)范化項為0.3等，具體可以設置的參數(shù)，可以通過explainParams()來獲取，還能看到程序已經(jīng)設置的參數(shù)的結果。>>>lr=LogisticRegression().\...setLabelCol("indexedLabel").\...setFeaturesCol("indexedFeatures").\...setMaxIter(100).\...setRegParam(0.3).\...setElasticNetParam(0.8)>>>print("LogisticRegressionparameters:\n"+lr.explainParams())8.4.1邏輯斯蒂回歸分類器第5步：設置一個IndexToString的轉(zhuǎn)換器，把預測的類別重新轉(zhuǎn)化成字符型的。構建一個機器學習流水線，設置各個階段。上一個階段的輸出將是本階段的輸入。>>>labelConverter=IndexToString().\...setInputCol("prediction").\...setOutputCol("predictedLabel").\...setLabels(labelIndexer.labels)>>>lrPipeline=Pipeline().\...setStages([labelIndexer,featureIndexer,lr,labelConverter])8.4.1邏輯斯蒂回歸分類器第6步：把數(shù)據(jù)集隨機分成訓練集和測試集，其中訓練集占70%。Pipeline本質(zhì)上是一個評估器，當Pipeline調(diào)用fit()的時候就產(chǎn)生了一個PipelineModel，它是一個轉(zhuǎn)換器。然后，這個PipelineModel就可以調(diào)用transform()來進行預測，生成一個新的DataFrame，即利用訓練得到的模型對測試集進行驗證。>>>trainingData,testData=data.randomSplit([0.7,0.3])>>>lrPipelineModel=lrPipeline.fit(trainingData)>>>lrPredictions=lrPipelineModel.transform(testData)8.4.1邏輯斯蒂回歸分類器第7步：輸出預測的結果，其中，select選擇要輸出的列，collect獲取所有行的數(shù)據(jù)，用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步：對訓練的模型進行評估。創(chuàng)建一個MulticlassClassificationEvaluator實例，用setter方法把預測分類的列名和真實分類的列名進行設置，然后計算預測準確率。>>>evaluator=MulticlassClassificationEvaluator().\...setLabelCol("indexedLabel").\...setPredictionCol("prediction")>>>lrAccuracy=evaluator.evaluate(lrPredictions)>>>lrAccuracy0.7774712643678161#模型預測的準確率8.4.1邏輯斯蒂回歸分類器第9步：可以通過model來獲取訓練得到的邏輯斯蒂模型。lrPipelineModel是一個PipelineModel，因此，可以通過調(diào)用它的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.11530503231364186,-0.63496556499483,0.750270597308472]numClasses:3numFeatures:48.4.2決策樹分類器決策樹（decisiontree）是一種基本的分類與回歸方法，這里主要介紹用于分類的決策樹。決策樹模式呈樹形結構，其中每個內(nèi)部節(jié)點表示一個屬性上的測試，每個分支代表一個測試輸出，每個葉節(jié)點代表一種類別。學習時利用訓練數(shù)據(jù)，根據(jù)損失函數(shù)最小化的原則建立決策樹模型；預測時，對新的數(shù)據(jù)，利用決策樹模型進行分類決策樹學習通常包括3個步驟：特征選擇、決策樹的生成和決策樹的剪枝8.4.2決策樹分類器我們以iris數(shù)據(jù)集（iris）為例進行分析（iris下載地址：/blog/wp-content/uploads/2017/03/iris.txt）iris以鳶尾花的特征作為數(shù)據(jù)來源，數(shù)據(jù)集包含150個數(shù)據(jù)集，分為3類，每類50個數(shù)據(jù)，每個數(shù)據(jù)包含4個屬性，是在數(shù)據(jù)挖掘、數(shù)據(jù)分類中非常常用的測試集、訓練集。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把每行的數(shù)據(jù)用“,”隔開，比如在我們的數(shù)據(jù)集中，每行被分成了5部分，前4部分是鳶尾花的4個特征，最后一部分是鳶尾花的分類；我們這里把特征存儲在Vector中，創(chuàng)建一個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決策樹分類器>>>labelIndexer=StringIndexer().\...setInputCol("label").\...setOutputCol("indexedLabel").\...fit(data)>>>featureIndexer=VectorIndexer().\...setInputCol("features").\...setOutputCol("indexedFeatures").\...setMaxCategories(4).\...fit(data)>>>labelConverter=IndexToString().\...setInputCol("prediction").\