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

本章內(nèi)容主要介紹了Spark生態(tài)系統(tǒng)中關(guān)于機(jī)器學(xué)習(xí)的ML庫(kù)。ML庫(kù)提供了一系列機(jī)器學(xué)習(xí)算法和工具，基于SparkDataFrame的API使得機(jī)器學(xué)習(xí)任務(wù)可以在大規(guī)模數(shù)據(jù)集上高效地執(zhí)行。本章首先介紹了機(jī)器學(xué)習(xí)的概念以及Spark機(jī)器學(xué)習(xí)庫(kù)ML，然后介紹了ML的基本數(shù)據(jù)類型、流水線和特征工程，以及回歸、分類、聚類和協(xié)同過濾算法等常用機(jī)器學(xué)習(xí)算法。通過學(xué)習(xí)本章，讀者將能夠深入了解ML，并根據(jù)實(shí)際需求選擇合適的算法和工具進(jìn)行處理。學(xué)習(xí)目標(biāo)011.了解機(jī)器學(xué)習(xí)和ML的基本概念022.掌握ML基本數(shù)據(jù)類型概念和創(chuàng)建044.理解ML特征工程處理055.理解ML各種常用機(jī)器學(xué)習(xí)算法使用場(chǎng)景和模型選擇033.掌握ML流水線的概念和創(chuàng)建學(xué)習(xí)目標(biāo)6.1Spark機(jī)器學(xué)習(xí)庫(kù)概述機(jī)器學(xué)習(xí)作為人工智能的核心驅(qū)動(dòng)力，正在以前所未有的速度重塑人類社會(huì)的各個(gè)方面。在大數(shù)據(jù)時(shí)代，大數(shù)據(jù)為機(jī)器學(xué)習(xí)提供了豐富的訓(xùn)練數(shù)據(jù)和測(cè)試數(shù)據(jù)，使得機(jī)器學(xué)習(xí)模型能夠不斷優(yōu)化和提升性能，如何將大數(shù)據(jù)和機(jī)器學(xué)習(xí)高效融合成為新的目標(biāo)。Spark立足于內(nèi)存計(jì)算，天然地適應(yīng)于迭代式計(jì)算，并且豐富的生態(tài)系統(tǒng)提供了一個(gè)基于海量數(shù)據(jù)的機(jī)器學(xué)習(xí)庫(kù)，它提供了常用機(jī)器學(xué)習(xí)算法的分布式實(shí)現(xiàn)。6.1.1機(jī)器學(xué)習(xí)簡(jiǎn)介機(jī)器學(xué)習(xí)是人工智能的一個(gè)重要分支，它致力于開發(fā)能夠從數(shù)據(jù)中自動(dòng)學(xué)習(xí)并改進(jìn)的算法和模型。機(jī)器學(xué)習(xí)系統(tǒng)通過分析大量的輸入數(shù)據(jù)來識(shí)別模式、提取特征，并據(jù)此構(gòu)建數(shù)學(xué)模型，從而能夠?qū)π聰?shù)據(jù)進(jìn)行預(yù)測(cè)或決策，這種能力使得機(jī)器學(xué)習(xí)在處理復(fù)雜且動(dòng)態(tài)變化的問題時(shí)尤其有效。6.1.1機(jī)器學(xué)習(xí)簡(jiǎn)介傳統(tǒng)的機(jī)器學(xué)習(xí)基于大數(shù)據(jù)的機(jī)器學(xué)習(xí)受限于較小規(guī)模的數(shù)據(jù)集有限的計(jì)算資源模型過擬合泛化能力不足對(duì)復(fù)雜模式識(shí)別局限利用海量數(shù)據(jù)

強(qiáng)大的分布式計(jì)算框架模型的準(zhǔn)確性和魯棒性提升發(fā)現(xiàn)更加細(xì)微和深層的數(shù)據(jù)模式預(yù)測(cè)、分類和決策支持更精準(zhǔn)6.1.2SparkMLlibMLlib是Spark中較舊的機(jī)器學(xué)習(xí)庫(kù)，它主要基于RDD（彈性分布式數(shù)據(jù)集）提供了一系列機(jī)器學(xué)習(xí)算法和工具，旨在簡(jiǎn)化機(jī)器學(xué)習(xí)的工程實(shí)踐工作，并方便擴(kuò)展到更大規(guī)模。MLlib由一些通用的學(xué)習(xí)算法和工具組成，包括分類、回歸、聚類、協(xié)同過濾等。特征化工具特征提取、轉(zhuǎn)化和選擇工具實(shí)用工具線性代數(shù)、統(tǒng)計(jì)、數(shù)據(jù)處理等工具持久性保存和加載算法、模型算法工具常用的學(xué)習(xí)算法，如分類、回歸、聚類和協(xié)同過濾010203046.1.3SparkML提供了基于DataFrame高層次的API，可以用來構(gòu)建機(jī)器學(xué)習(xí)流水線。ML

Pipeline彌補(bǔ)了原始MLlib庫(kù)的不足，向用戶提供了一個(gè)基于DataFrame的機(jī)器學(xué)習(xí)流水線式API套件。Spark.ML包含基于RDD的原始算法API。SparkMLlib歷史比較長(zhǎng)，在1.0以前的版本即已經(jīng)包含了，提供的算法實(shí)現(xiàn)都是基于原始的RDD。Spark.MLlibSpark機(jī)器學(xué)習(xí)庫(kù)從1.2版本以后被分為Spark.MLlib和Spark.ML兩個(gè)包，Spark官方推薦使用Spark.ML。開發(fā)者需要注意的是，從Spark2.0開始，基于RDD的MLlibAPI進(jìn)入維護(hù)模式，不再增加任何新的特性。6.2基本數(shù)據(jù)類型基本數(shù)據(jù)類型標(biāo)簽點(diǎn)（LabeledPoint）本地矩陣（LocalMatrix）本地向量（LocalVector）6.2.1本地向量本地向量存儲(chǔ)在單機(jī)上,ML提供了兩種類型的本地向量，稠密向量（DenseVector）和稀疏向量（SparseVector）。稠密向量使用一個(gè)Double類型數(shù)組來表示其中每一維元素，而稀疏向量則是基于一個(gè)Int類型索引數(shù)組和一個(gè)Double類型的值數(shù)組。例如，向量(1.0,0.0,3.0)的稠密向量表示形式是[1.0,0.0,3.0]，而稀疏向量形式則是(3,[0,2],[1.0,3.0])，其中，3是向量的長(zhǎng)度，[0,2]是向量中非0維度的索引值，表示位置為0、2的兩個(gè)元素為非零值，而[1.0,3.0]則是按索引排列的數(shù)組元素值。6.2.1本地向量importorg.apache.spark.ml.linalg.{Vector,Vectors}objectVectorExample{defmain(args:Array[String]):Unit={//創(chuàng)建一個(gè)稠密向量valdv:Vector=Vectors.dense(Array(1.0,2.0,3.0))println(s"DenseVector:$dv")//創(chuàng)建一個(gè)稀疏向量//參數(shù)順序是：向量的大小，非零元素的索引列表，非零元素的值列表valsv:Vector=Vectors.sparse(3,Array(0,2),Array(1.0,3.0))println(s"SparseVector:$sv")}}Scala所有本地向量都以org.apache.spark.ml.linalg.Vector為基類，DenseVector和SparseVector分別是它的兩個(gè)實(shí)現(xiàn)類，故推薦使用Vectors工具類下定義的工廠方法來創(chuàng)建本地向量,需要注意的是程序會(huì)默認(rèn)引入scala.collection.immutable.Vector，需要顯式地引入org.apache.spark.ml.linalg.Vector來使用ML提供的向量類型。6.2.1本地向量的創(chuàng)建frompyspark.ml.linalgimportVectorsdefmain():#創(chuàng)建一個(gè)稠密向量dv=Vectors.dense([1.0,2.0,3.0])print(f"DenseVector:{dv}")#創(chuàng)建一個(gè)稀疏向量#參數(shù)順序是：向量的大小，非零元素的索引列表，非零元素的值列表sv=Vectors.sparse(3,[0,2],[1.0,3.0])print(f"SparseVector:{sv}")if__name__=="__main__":main()Python運(yùn)行結(jié)果：6.2.2標(biāo)簽點(diǎn)標(biāo)簽點(diǎn)是一種帶有標(biāo)簽的本地向量，它可以是稠密或者是稀疏的。在ML中，標(biāo)注點(diǎn)在監(jiān)督學(xué)習(xí)算法中被使用。由于標(biāo)簽是用Double類型來存儲(chǔ)的，所以標(biāo)簽點(diǎn)類型在回歸和分類問題上均可使用。例如，對(duì)于二分類問題，則正樣本的標(biāo)簽為1，負(fù)樣本的標(biāo)簽為0，而對(duì)于多類別的分類問題來說，標(biāo)簽則應(yīng)是一個(gè)以0開始的索引序列:0,1,2...注：在PySpark的ML庫(kù)中，沒有LabeledPoint這個(gè)類，監(jiān)督學(xué)習(xí)的數(shù)據(jù)通常以DataFrame的形式存在，DataFrame包含兩列：label和features。label列包含監(jiān)督學(xué)習(xí)的目標(biāo)值，而features列包含一個(gè)向量，該向量包含了用于模型訓(xùn)練的特征值。在新的PySparkAPI設(shè)計(jì)下，不需要專門的LabeledPoint類，而是直接使用DataFrame來組織和處理數(shù)據(jù)。6.2.2標(biāo)簽點(diǎn)importorg.apache.spark.ml.linalg.Vectorsimportorg.apache.spark.ml.feature.LabeledPointobjectLabeledPointExample{defmain(args:Array[String]):Unit={//創(chuàng)建一個(gè)標(biāo)簽為1.0（分類中可視為正樣本）的稠密向量標(biāo)簽點(diǎn)valpos:LabeledPoint=LabeledPoint(1.0,Vectors.dense(Array(1.0,2.0,3.0)))println(s"PostiveLabel:$pos")//創(chuàng)建一個(gè)標(biāo)簽為0.0（分類中可視為負(fù)樣本）的稀疏向量標(biāo)簽點(diǎn)valneg:LabeledPoint=LabeledPoint(0.0,Vectors.sparse(3,Array(0,2),Array(1.0,3.0)))println(s"NegativeLabel:$neg")}}Scala標(biāo)簽點(diǎn)的實(shí)現(xiàn)類是org.apache.spark.ml.feature.LabeledPoint，請(qǐng)注意它與前面介紹的本地向量不同，并不位于linalg包下。運(yùn)行結(jié)果：6.2.2標(biāo)簽點(diǎn)需要注意的是，在PySpark的ML庫(kù)中，沒有LabeledPoint這個(gè)類，監(jiān)督學(xué)習(xí)的數(shù)據(jù)通常以DataFrame的形式存在，DataFrame包含兩列：label和features。label列包含監(jiān)督學(xué)習(xí)的目標(biāo)值，而features列包含一個(gè)向量，該向量包含了用于模型訓(xùn)練的特征值。在新的PySparkAPI設(shè)計(jì)下，不需要專門的LabeledPoint類，而是直接使用DataFrame來組織和處理數(shù)據(jù)。標(biāo)簽點(diǎn)的創(chuàng)建只提供Scala語(yǔ)言的代碼。6.2.3本地矩陣本地矩陣存儲(chǔ)在單機(jī)上，具有Int類型的行、列索引值和Double類型的元素值。ML支持稠密矩陣（DenseMatrix）和稀疏矩陣（SparseMatrix）兩種本地矩陣，稠密矩陣將所有元素的值存儲(chǔ)在一個(gè)列優(yōu)先的Double類數(shù)組中，而稀疏矩陣則將非零元素以列優(yōu)先的壓縮稀疏列CSC（CompressedSparseColumn）模式進(jìn)行存儲(chǔ)。對(duì)于矩陣，使用程序創(chuàng)建稠密矩陣和稀疏矩陣。6.2.3本地矩陣importorg.apache.spark.ml.linalg.{Matrix,Matrices}objectMatrixExample{defmain(args:Array[String]):Unit={//創(chuàng)建一個(gè)4x3的稠密矩陣((1.0,0.0,0.0),(0.0,2.0,0.0),(0.0,0.0,3.0),(4.0,0.0,0.0))，按列優(yōu)先順序逐列填充矩陣valdm:Matrix=Matrices.dense(4,3,Array(1.0,0.0,0.0,0.0,2.0,0.0,0.0,0.0,3.0,4.0,0.0,0.0))println(s"DenseMatrix:\n$dm")//創(chuàng)建一個(gè)4x3的稀疏矩陣((1.0,0.0,0.0),(0.0,2.0,0.0),(0.0,0.0,3.0),(4.0,0.0,0.0))valsm:Matrix=Matrices.sparse(4,3,Array(0,2,3,4),Array(0,3,1,2),Array(1.0,4.0,2.0,3.0))println(s"SparseMatrix:\n$sm")}}Scala6.2.3本地矩陣frompyspark.ml.linalgimportMatricesdefmain():#創(chuàng)建一個(gè)4x3的稠密矩陣((1.0,0.0,0.0),(0.0,2.0,0.0),(0.0,0.0,3.0),(4.0,0.0,0.0)),按列優(yōu)先順序逐列填充矩陣dm=Matrices.dense(4,3,[1.0,0.0,0.0,0.0,2.0,0.0,0.0,0.0,3.0,4.0,0.0,0.0])print(f"DenseMatrix:\n{dm}")#創(chuàng)建一個(gè)4x3的稀疏矩陣((1.0,0.0,0.0),(0.0,2.0,0.0),(0.0,0.0,3.0),(4.0,0.0,0.0))sm=Matrices.sparse(4,3,[0,2,3,4],[0,3,1,2],[1.0,4.0,2.0,3.0])print(f"SparseMatrix:\n{sm}")if__name__=="__main__":main()Python6.2.3本地矩陣對(duì)于矩陣，使用程序創(chuàng)建稠密矩陣和稀疏矩陣。運(yùn)行結(jié)果：6.2.4數(shù)據(jù)源ML支持多種數(shù)據(jù)源格式，可以方便地從不同的存儲(chǔ)系統(tǒng)中讀取數(shù)據(jù)進(jìn)行處理。下面將介紹一些特定的數(shù)據(jù)源，包括圖像數(shù)據(jù)源和LIBSSVM數(shù)據(jù)源。1.圖像數(shù)據(jù)源圖像數(shù)據(jù)源用于從目錄中加載圖像文件，它可以通過Java庫(kù)中的ImageIO加載壓縮的圖像（如JPEG、PNG等格式）到原始圖像。加載的DataFrame有一個(gè)StructType列image，其中包含以圖像模式存儲(chǔ)的圖像數(shù)據(jù)。6.2.4數(shù)據(jù)源image列的屬性如下：StringType表示圖像文件的路徑originIntegerType表示圖像的高度heightIntegerType表示圖像的寬度widthIntegerType表示圖像通道的數(shù)量nChannelsIntegerType表示與OpenCV兼容的類型modeBinaryType表示以O(shè)penCV兼容順序排列的圖像字節(jié)data6.2.4數(shù)據(jù)源2.LIBSVM數(shù)據(jù)源LIBSVM數(shù)據(jù)源用于從目錄加載libsvm類型的文件，通過LibSVMDataSource類將LIBSVM格式加載后生成DataFrame。LIBSVM起初是一個(gè)SVM模式識(shí)別與回歸的開源機(jī)器學(xué)習(xí)庫(kù)，主要提供有關(guān)支持向量機(jī)的算法，后提供了多種編程語(yǔ)言的接口，后經(jīng)過不斷改進(jìn)與其他機(jī)器學(xué)習(xí)技術(shù)融合，以解決更復(fù)雜的問題。6.2.4數(shù)據(jù)源通過LIBSVM數(shù)據(jù)源讀取的DataFrame有兩列：包含以Double類型存儲(chǔ)的標(biāo)簽和包含以Vectors類型存儲(chǔ)的特征向量的特征；每一行代表一條記錄，同時(shí)也是一個(gè)特征向量，格式為:<label1><index1>:<value1><index2>:<value2>..<label1>就是該記錄的標(biāo)簽；<index>代表索引；<value>是對(duì)應(yīng)位置的特征的屬性值。6.2.4數(shù)據(jù)源importorg.apache.spark.sql.SparkSession//定義一個(gè)Scala對(duì)象ImageDataExampleobjectImageDataExample{defmain(args:Array[String]):Unit={valspark=SparkSession.builder.appName("ImageDataExample").master("local").getOrCreate()//讀取圖像數(shù)據(jù)，使用"image"格式加載圖像文件//選項(xiàng)"dropInvalid"設(shè)置為true，表示在讀取過程中丟棄無效的圖像文件valpicdf=spark.read.format("image").option("dropInvalid",true).load("data/mllib/images/origin/kittens")//指定圖像文件的路徑//顯示圖像數(shù)據(jù)中的"origin"、"width"和"height"字段//truncate=false表示不截?cái)噍敵鰞?nèi)容picdf.select("image.origin","image.width","image.height").show(truncate=false)Scala6.2.4數(shù)據(jù)源importorg.apache.spark.sql.SparkSession//定義一個(gè)Scala對(duì)象ImageDataExampleobjectImageDataExample{defmain(args:Array[String]):Unit={valspark=SparkSession.builder.appName("ImageDataExample").master("local").getOrCreate()//讀取圖像數(shù)據(jù)，使用"image"格式加載圖像文件//選項(xiàng)"dropInvalid"設(shè)置為true，表示在讀取過程中丟棄無效的圖像文件valpicdf=spark.read.format("image").option("dropInvalid",true).load("data/mllib/images/origin/kittens")//指定圖像文件的路徑//顯示圖像數(shù)據(jù)中的"origin"、"width"和"height"字段//truncate=false表示不截?cái)噍敵鰞?nèi)容picdf.select("image.origin","image.width","image.height").show(truncate=false)Scala6.2.4數(shù)據(jù)源//讀取LibSVM格式的數(shù)據(jù)，使用"libsvm"格式加載數(shù)據(jù)//選項(xiàng)"numFeatures"設(shè)置為780，表示每個(gè)樣本有780個(gè)特征vallibsvmdf=spark.read.format("libsvm").option("numFeatures","780").load("data/mllib/sample_libsvm_data.txt")//指定LibSVM數(shù)據(jù)文件的路徑//顯示LibSVM數(shù)據(jù)的前10條記錄libsvmdf.show(10)//停止SparkSession，釋放資源spark.stop()}}Scala6.2.4數(shù)據(jù)源frompyspark.sqlimportSparkSessiondefmain():#創(chuàng)建SparkSession，這是與Spark交互的入口點(diǎn)spark=SparkSession\.builder\.appName("ImageDataExample")#設(shè)置應(yīng)用名稱.master("local")#設(shè)置運(yùn)行模式為本地模式（單機(jī)）.getOrCreate()#獲取或創(chuàng)建SparkSession實(shí)例#讀取圖像數(shù)據(jù)，使用"image"格式加載圖像文件#選項(xiàng)"dropInvalid"設(shè)置為True，表示在讀取過程中丟棄無效的圖像文件picdf=spark.read.format("image")\.option("dropInvalid",True)\.load("data/mllib/images/origin/kittens")#指定圖像文件的路徑#顯示圖像數(shù)據(jù)中的"origin"、"width"和"height"字段#truncate=False表示不截?cái)噍敵鰞?nèi)容picdf.select("image.origin","image.width","image.height").show(truncate=False)Python6.2.4數(shù)據(jù)源#讀取LibSVM格式的數(shù)據(jù)，使用"libsvm"格式加載數(shù)據(jù)#選項(xiàng)"numFeatures"設(shè)置為780，表示每個(gè)樣本有780個(gè)特征libsvmdf=(spark.read.format("libsvm").option("numFeatures","780").load("data/mllib/sample_libsvm_data.txt"))#指定LibSVM數(shù)據(jù)文件的路徑#顯示LibSVM數(shù)據(jù)的前10條記錄libsvmdf.show(10)#停止SparkSession，釋放資源spark.stop()if__name__=="__main__":main()Python6.2.4數(shù)據(jù)源運(yùn)行結(jié)果：6.3機(jī)器學(xué)習(xí)流水線機(jī)器學(xué)習(xí)構(gòu)建的過程通常包括源數(shù)據(jù)ETL、數(shù)據(jù)預(yù)處理、特征選取和模型訓(xùn)練與驗(yàn)證四個(gè)步驟，以上四個(gè)步驟可以抽象為一個(gè)包括多個(gè)步驟的流水線式工作，從數(shù)據(jù)收集開始至輸出需要的最終結(jié)果。因此，對(duì)多個(gè)步驟、進(jìn)行抽象建模，簡(jiǎn)化為流水線式的機(jī)器學(xué)習(xí)工作流程則更加高效、易用。本節(jié)將介紹機(jī)器學(xué)習(xí)流水線的概念及工作過程。6.3.1轉(zhuǎn)換器和評(píng)估器受到scikit-learn項(xiàng)目的啟發(fā)，并且總結(jié)了MLlib在處理復(fù)雜機(jī)器學(xué)習(xí)問題中工作繁雜、流程不清晰等弊端，ML旨在向用戶提供基于DataFrame之上的更加高層次的API庫(kù)，以更加方便的構(gòu)建復(fù)雜的機(jī)器學(xué)習(xí)工作流式應(yīng)用。MLPipeline流水線通常包含以下幾個(gè)重要概念：DataFrameTransformer轉(zhuǎn)換器Pipeline流水線Estimator評(píng)估器Parameter參數(shù)6.3.1轉(zhuǎn)換器和評(píng)估器使用SparkSQL中的DataFrame作為數(shù)據(jù)集，支持多種數(shù)據(jù)類型。與RDD相比，DataFrame內(nèi)嵌了schema信息，更貼近于傳統(tǒng)數(shù)據(jù)庫(kù)的表格結(jié)構(gòu)。它用于MLPipeline中以存儲(chǔ)原始數(shù)據(jù)。例如，DataFrame的列可以存儲(chǔ)文本、特征向量、實(shí)際標(biāo)簽和預(yù)測(cè)結(jié)果等。DataFrame16.3.1轉(zhuǎn)換器和評(píng)估器轉(zhuǎn)換器是一種抽象，包括特征轉(zhuǎn)換器和學(xué)習(xí)模型。轉(zhuǎn)換器通過transform()方法，該方法將一個(gè)DataFrame轉(zhuǎn)換為另一個(gè)DataFrame，通常通過添加一個(gè)或多個(gè)列。例如，一個(gè)特征轉(zhuǎn)換器可能會(huì)接收一個(gè)DataFrame，讀取一列（例如文本），將其映射到一個(gè)新列（例如特征向量），最終生成一個(gè)包含了這個(gè)新列的DataFrame。一個(gè)學(xué)習(xí)模型可能會(huì)接收一個(gè)DataFrame，讀取包含特征向量的列，為每個(gè)特征向量預(yù)測(cè)標(biāo)簽，并輸出一個(gè)將預(yù)測(cè)標(biāo)簽作為列附加的新DataFrame。轉(zhuǎn)換器26.3.1轉(zhuǎn)換器和評(píng)估器評(píng)估器是機(jī)器學(xué)習(xí)算法或在訓(xùn)練數(shù)據(jù)上的訓(xùn)練方法的概念抽象。在流水線中通常用來被操作DataFrame數(shù)據(jù)并生產(chǎn)一個(gè)轉(zhuǎn)換器。評(píng)估器實(shí)現(xiàn)了方法fit（），它接受一個(gè)DataFrame并產(chǎn)生一個(gè)轉(zhuǎn)換器。例如，邏輯斯蒂回歸算法LogisticRegression就是一個(gè)評(píng)估器，通過調(diào)用fit（）方法訓(xùn)練特征數(shù)據(jù)而得到一個(gè)邏輯斯蒂回模型，該學(xué)習(xí)模型就是一個(gè)新生成的轉(zhuǎn)換器。評(píng)估器36.3.1轉(zhuǎn)換器和評(píng)估器流水線又稱管道，將多個(gè)工作流階段（轉(zhuǎn)換器和評(píng)估器）連接在一起，形成機(jī)器學(xué)習(xí)的工作流，并獲得結(jié)果輸出。具體工作流程將在6.3.2部分詳細(xì)介紹。流水線4參數(shù)被用來設(shè)置轉(zhuǎn)換器或者評(píng)估器，所有轉(zhuǎn)換器和評(píng)估器可共享用于指定參數(shù)的公共API。參數(shù)56.3.1轉(zhuǎn)換器和評(píng)估器importorg.apache.spark.sql.SparkSession//導(dǎo)入SparkSession，用于創(chuàng)建Spark應(yīng)用importorg.apache.spark.ml.classification.LogisticRegression//導(dǎo)入邏輯回歸模型類importorg.apache.spark.ml.linalg.{Vector,Vectors}//導(dǎo)入向量類和向量工廠類importorg.apache.spark.ml.param.ParamMap//導(dǎo)入?yún)?shù)映射類importorg.apache.spark.sql.Row//導(dǎo)入行類，用于處理DataFrame的行objectEstimatorandTransformerExample{defmain(args:Array[String]):Unit={valspark=SparkSession.builder.appName("EstimatorandTransformerExample").master("local").getOrCreate()//準(zhǔn)備訓(xùn)練數(shù)據(jù)，創(chuàng)建一個(gè)包含標(biāo)簽和特征的DataFramevaltraining=spark.createDataFrame(Seq((1.0,Vectors.dense(0.0,1.1,0.1)),(0.0,Vectors.dense(2.0,1.0,-1.0)),(0.0,Vectors.dense(2.0,1.3,1.0)),(1.0,Vectors.dense(0.0,1.2,-0.5)))).toDF("label","features")Scala6.3.1轉(zhuǎn)換器和評(píng)估器//創(chuàng)建一個(gè)邏輯回歸模型實(shí)例，這是一個(gè)評(píng)估器vallr=newLogisticRegression()//打印邏輯回歸模型的參數(shù)，包括文檔和默認(rèn)值println(s"LogisticRegressionparameters:\n${lr.explainParams()}\n")//使用setter方法設(shè)置參數(shù)lr.setMaxIter(10)//設(shè)置最大迭代次數(shù)為10.setRegParam(0.01)//設(shè)置正則化參數(shù)為0.01//使用訓(xùn)練數(shù)據(jù)擬合模型，返回一個(gè)模型（轉(zhuǎn)換器）valmodel1=lr.fit(training)//打印模型1在訓(xùn)練時(shí)使用的參數(shù)println(s"Model1wasfitusingparameters:${model1.parent.extractParamMap}")//使用ParamMap設(shè)置參數(shù)valparamMap=ParamMap(lr.maxIter->20).put(lr.maxIter,30)//覆蓋原始最大迭代次數(shù)為30.put(lr.regParam->0.1,lr.threshold->0.55)//正則化參數(shù)為0.1，閾值為0.55Scala6.3.1轉(zhuǎn)換器和評(píng)估器//創(chuàng)建另一個(gè)ParamMap，用于更改輸出列的名稱valparamMap2=ParamMap(babilityCol->"myProbability")//合并ParamMapvalparamMapCombined=paramMap++paramMap2//使用合并后的ParamMap訓(xùn)練新的模型valmodel2=lr.fit(training,paramMapCombined)println(s"Model2wasfitusingparameters:${model2.parent.extractParamMap}")//準(zhǔn)備測(cè)試數(shù)據(jù)valtest=spark.createDataFrame(Seq((1.0,Vectors.dense(-1.0,1.5,1.3)),(0.0,Vectors.dense(3.0,2.0,-0.1)),(1.0,Vectors.dense(0.0,2.2,-1.5)))).toDF("label","features")Scala6.3.1轉(zhuǎn)換器和評(píng)估器//使用模型2對(duì)測(cè)試數(shù)據(jù)進(jìn)行預(yù)測(cè)model2.transform(test).select("features","label","myProbability","prediction")//選擇需要的列.collect()//收集結(jié)果.foreach{caseRow(features:Vector,label:Double,prob:Vector,prediction:Double)=>println(s"($features,$label)->prob=$prob,prediction=$prediction")//打印每個(gè)樣本的預(yù)測(cè)結(jié)果}spark.stop()//停止SparkSession}}Scala6.3.1轉(zhuǎn)換器和評(píng)估器frompyspark.sqlimportSparkSession#導(dǎo)入SparkSession，用于創(chuàng)建Spark應(yīng)用frompyspark.ml.linalgimportVectors#用于創(chuàng)建特征向量frompyspark.ml.classificationimportLogisticRegression#用于創(chuàng)建邏輯回歸模型defmain():#創(chuàng)建一個(gè)SparkSession對(duì)象，這是使用Spark的入口點(diǎn)spark=SparkSession\.builder\.appName("EstimatorandTransformerExample")\.master("local")\.getOrCreate()#準(zhǔn)備訓(xùn)練數(shù)據(jù)，創(chuàng)建一個(gè)包含標(biāo)簽和特征向量的DataFrametraining=spark.createDataFrame([(1.0,Vectors.dense([0.0,1.1,0.1])),(0.0,Vectors.dense([2.0,1.0,-1.0])),(0.0,Vectors.dense([2.0,1.3,1.0])),(1.0,Vectors.dense([0.0,1.2,-0.5]))],["label","features"])Python6.3.1轉(zhuǎn)換器和評(píng)估器#創(chuàng)建一個(gè)邏輯回歸模型實(shí)例，這是一個(gè)評(píng)估器，設(shè)置最大迭代次數(shù)為10，設(shè)置正則化參數(shù)為0.01lr=LogisticRegression(maxIter=10,regParam=0.01)#打印邏輯回歸模型的參數(shù)，包括文檔和默認(rèn)值print("LogisticRegressionparameters:\n"+lr.explainParams()+"\n")#使用訓(xùn)練數(shù)據(jù)擬合模型，返回一個(gè)模型（轉(zhuǎn)換器）model1=lr.fit(training)#打印模型1在訓(xùn)練時(shí)使用的參數(shù)print("Model1wasfitusingparameters:")print(model1.extractParamMap())#創(chuàng)建一個(gè)參數(shù)映射paramMap={lr.maxIter:20}#覆蓋原始最大迭代次數(shù)為30paramMap[lr.maxIter]=30#正則化參數(shù)為0.1，閾值為0.55paramMap.update({lr.regParam:0.1,lr.threshold:0.55})

Python6.3.1轉(zhuǎn)換器和評(píng)估器#創(chuàng)建另一個(gè)參數(shù)映射，用于更改輸出列的名稱paramMap2={babilityCol:"myProbability"}#復(fù)制參數(shù)映射paramMapCombined=paramMap.copy()#合并參數(shù)映射paramMapCombined.update(paramMap2)#使用合并后的參數(shù)映射訓(xùn)練新的模型model2=lr.fit(training,paramMapCombined)print("Model2wasfitusingparameters:")print(model2.extractParamMap())#準(zhǔn)備測(cè)試數(shù)據(jù)test=spark.createDataFrame([(1.0,Vectors.dense([-1.0,1.5,1.3])),(0.0,Vectors.dense([3.0,2.0,-0.1])),(1.0,Vectors.dense([0.0,2.2,-1.5]))],["label","features"])

Python6.3.1轉(zhuǎn)換器和評(píng)估器#使用模型2對(duì)測(cè)試數(shù)據(jù)進(jìn)行預(yù)測(cè)prediction=model2.transform(test)#收集結(jié)果result=prediction.select("features","label","myProbability","prediction")\.collect()#打印預(yù)測(cè)結(jié)果forrowinresult:print("features=%s,label=%s->prob=%s,prediction=%s"%(row.features,row.label,row.myProbability,row.prediction))#停止SparkSessionspark.stop()if__name__=="__main__":main()Python6.3.1轉(zhuǎn)換器和評(píng)估器運(yùn)行結(jié)果第一部分（截圖顯示不完整）輸出列出了邏輯回歸模型的所有參數(shù)及其默認(rèn)值或當(dāng)前值6.3.1轉(zhuǎn)換器和評(píng)估器運(yùn)行結(jié)果第二部分為模型1和模型2的參數(shù)，輸出顯示了兩個(gè)模型在訓(xùn)練時(shí)使用的參數(shù)。6.3.1轉(zhuǎn)換器和評(píng)估器第三部分輸出顯示了模型2對(duì)測(cè)試數(shù)據(jù)集的預(yù)測(cè)結(jié)果。每行顯示一個(gè)測(cè)試樣本的特征、實(shí)際標(biāo)簽、預(yù)測(cè)概率和預(yù)測(cè)類別。在以上案例中，通過LogisticRegression類創(chuàng)建的lr就是一個(gè)評(píng)估器，lr通過fit方法訓(xùn)練模型，得到新的轉(zhuǎn)換器model1和model2；model2通過transform方法被用來將測(cè)試數(shù)據(jù)集test轉(zhuǎn)換成包含預(yù)測(cè)結(jié)果的新DataFrame。選擇特定的列并收集結(jié)果，然后打印每個(gè)樣本的特征、標(biāo)簽、概率和預(yù)測(cè)值，該過程就是一個(gè)轉(zhuǎn)換器。轉(zhuǎn)換器和評(píng)估器使用統(tǒng)一的API指定參數(shù)，既可以直接設(shè)置實(shí)例的參數(shù)，也可以通過傳遞ParamMap給fit（）或transform（）的方式。6.3.2流水線要構(gòu)建一個(gè)流水線，首先需要定義流水線中的各個(gè)流水線階段（PipelineStage），各階段包括各種轉(zhuǎn)換器和評(píng)估器完成特定的任務(wù)。上一個(gè)流水線階段的輸出，可以作為下一個(gè)流水線階段的輸入。這些階段按順序執(zhí)行，輸入的DataFrame在通過每個(gè)階段時(shí)被轉(zhuǎn)換。在轉(zhuǎn)換器階段，在DataFrame上調(diào)用transform()方法。在評(píng)估器階段，調(diào)用fit()方法以產(chǎn)生一個(gè)轉(zhuǎn)換器，然后在DataFrame上調(diào)用該轉(zhuǎn)換器的transform()方法。6.3.2流水線調(diào)用transform（）方法將原始文本拆分為單詞，向DataFrame添加一個(gè)帶有單詞的新列；Tokenizer階段調(diào)用transform（）方法將DataFrame中的單詞列轉(zhuǎn)換成特征向量，把這些向量添加一個(gè)新列到DataFrame；HashingTF階段評(píng)估器通過調(diào)用fit（）方法，產(chǎn)生LogisticRegressionModel。如果流水線后續(xù)仍有階段，則會(huì)將DataFrame傳遞到下階段前，調(diào)用LogisticRegressionModel的transform（）方法。LogisticRegression階段6.3.2流水線流水線本身就是一個(gè)評(píng)估器，所以在流水線的fit（）方法調(diào)用后，會(huì)產(chǎn)生一個(gè)流水線模型（PipelineModel），同時(shí)流水線模型也是一個(gè)轉(zhuǎn)換器，可用于數(shù)據(jù)測(cè)試。流水線模型具有與原流水線相同的階段數(shù)，但是原流水線中的所有評(píng)估器都變?yōu)檗D(zhuǎn)換器。當(dāng)在測(cè)試數(shù)據(jù)集上調(diào)用流水線模型的transform（）方法時(shí)，測(cè)試數(shù)據(jù)按順序通過流水線的各階段。每個(gè)階段的transform（）方法更新DataFrame并將其傳遞到下一個(gè)階段。通過這種方式，可以保證流水線和流水線模型對(duì)于訓(xùn)練數(shù)據(jù)和測(cè)試數(shù)據(jù)都通過相同的特征處理步驟。6.3.2流水線importorg.apache.spark.sql.SparkSession//導(dǎo)入SparkSession，用于創(chuàng)建Spark應(yīng)用importorg.apache.spark.ml.{Pipeline,PipelineModel}//導(dǎo)入流水線和流水線模型類importorg.apache.spark.ml.classification.LogisticRegression//導(dǎo)入邏輯斯蒂回歸模型類importorg.apache.spark.ml.feature.{HashingTF,Tokenizer}//導(dǎo)入Tokenizer和HashingTF類importorg.apache.spark.ml.linalg.Vector//導(dǎo)入向量類importorg.apache.spark.sql.Row//導(dǎo)入行類，用于處理DataFrame的行objectPipelineExample{defmain(args:Array[String]):Unit={valspark=SparkSession.builder.appName("PipelineExample").master("local").getOrCreate()//準(zhǔn)備訓(xùn)練數(shù)據(jù)，創(chuàng)建一個(gè)包含ID、文本和標(biāo)簽的DataFramevaltraining=spark.createDataFrame(Seq((0L,"abcdespark",1.0),(1L,"bd",0.0),(2L,"sparkfgh",1.0),(3L,"hadoopmapreduce",0.0))).toDF("id","text","label")Scala6.3.2流水線//配置機(jī)器學(xué)習(xí)管道，包括三個(gè)階段：tokenizer、hashingTF和lrvaltokenizer=newTokenizer().setInputCol("text")//設(shè)置輸入列.setOutputCol("words")//設(shè)置輸出列valhashingTF=newHashingTF().setNumFeatures(1000)//設(shè)置特征數(shù)量.setInputCol(tokenizer.getOutputCol)//設(shè)置輸入列.setOutputCol("features")//設(shè)置輸出列vallr=newLogisticRegression().setMaxIter(10)//設(shè)置最大迭代次數(shù).setRegParam(0.001)//設(shè)置正則化參數(shù)valpipeline=newPipeline().setStages(Array(tokenizer,hashingTF,lr))//設(shè)置管道的階段//使用訓(xùn)練數(shù)據(jù)擬合管道valmodel=pipeline.fit(training)Scala6.3.2流水線//將擬合好的管道模型保存到磁盤model.write.overwrite().save("/tmp/spark-logistic-regression-model")//保存未訓(xùn)練的管道到磁盤pipeline.write.overwrite().save("/tmp/unfit-lr-model")//在生產(chǎn)環(huán)境中加載模型valsameModel=PipelineModel.load("/tmp/spark-logistic-regression-model")//準(zhǔn)備測(cè)試數(shù)據(jù)，創(chuàng)建一個(gè)包含ID和文本的DataFramevaltest=spark.createDataFrame(Seq((4L,"sparkijk"),(5L,"lmn"),(6L,"sparkhadoopspark"),(7L,"apachehadoop"))).toDF("id","text")Scala6.3.2流水線//使用模型對(duì)測(cè)試數(shù)據(jù)進(jìn)行預(yù)測(cè)model.transform(test).select("id","text","probability","prediction")//選擇需要的列.collect()//收集結(jié)果.foreach{caseRow(id:Long,text:String,prob:Vector,prediction:Double)=>println(s"($id,$text)-->prob=$prob,prediction=$prediction")//打印每個(gè)文檔的預(yù)測(cè)結(jié)果}spark.stop()//停止SparkSession}}Scala6.3.2流水線frompyspark.sqlimportSparkSession#導(dǎo)入SparkSession，用于創(chuàng)建Spark應(yīng)用frompyspark.mlimportPipeline#導(dǎo)入流水線類frompyspark.ml.classificationimportLogisticRegression#導(dǎo)入邏輯斯蒂回歸模型類frompyspark.ml.featureimportHashingTF,Tokenizer#導(dǎo)入Tokenizer和HashingTF類defmain():spark=SparkSession\.builder\.appName("EstimatorandTransformerExample")\#設(shè)置應(yīng)用名稱.master("local")\#設(shè)置運(yùn)行模式為本地模式.getOrCreate()#創(chuàng)建或獲取一個(gè)SparkSession對(duì)象#準(zhǔn)備訓(xùn)練數(shù)據(jù)，創(chuàng)建一個(gè)包含ID、文本和標(biāo)簽的DataFrametraining=spark.createDataFrame([(0,"abcdespark",1.0),(1,"bd",0.0),(2,"sparkfgh",1.0),(3,"hadoopmapreduce",0.0)],["id","text","label"])Python6.3.2流水線#配置機(jī)器學(xué)習(xí)管道，包括三個(gè)階段：tokenizer、hashingTF和lrtokenizer=Tokenizer(inputCol="text",outputCol="words")#創(chuàng)建分詞器hashingTF=HashingTF(inputCol=tokenizer.getOutputCol(),outputCol="features")#創(chuàng)建哈希技術(shù)特征轉(zhuǎn)換器lr=LogisticRegression(maxIter=10,regParam=0.001)#創(chuàng)建邏輯回歸分類器pipeline=Pipeline(stages=[tokenizer,hashingTF,lr])#創(chuàng)建管道#使用訓(xùn)練數(shù)據(jù)擬合管道m(xù)odel=pipeline.fit(training)#準(zhǔn)備測(cè)試數(shù)據(jù)，創(chuàng)建一個(gè)包含ID和文本的DataFrametest=spark.createDataFrame([(4,"sparkijk"),(5,"lmn"),(6,"sparkhadoopspark"),(7,"apachehadoop")],["id","text"])Python6.3.2流水線#使用模型對(duì)測(cè)試數(shù)據(jù)進(jìn)行預(yù)測(cè)，并打印預(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#type:ignore))spark.stop()#停止SparkSessionif__name__=="__main__":main()Python6.3.2流水線運(yùn)行結(jié)果：述代碼展示了如何構(gòu)建和使用一個(gè)簡(jiǎn)單的文本分類流水線，包括文本預(yù)處理和模型訓(xùn)練，模型經(jīng)過訓(xùn)練能夠根據(jù)文本內(nèi)容正確地預(yù)測(cè)標(biāo)簽，對(duì)于包含hadoop和mapreduce的文本，模型傾向于預(yù)測(cè)為負(fù)類（標(biāo)簽0）；對(duì)于包含spark的文本，模型更可能預(yù)測(cè)為正類（標(biāo)簽1），除非文本中還包含了其他特征詞，如hadoop。通過這種方式，可以輕松地將多個(gè)步驟組合在一起，形成一個(gè)端到端的機(jī)器學(xué)習(xí)流水線。6.4特征工程特征工程就是指用一系列工程化的方式從原始數(shù)據(jù)中篩選出更好的數(shù)據(jù)特征，以提升模型的訓(xùn)練效果。簡(jiǎn)而言之，就是為算法提供更友好的輸入，以最大化地發(fā)揮算法的作用。同時(shí)，特征工程雖然是一種技術(shù)，但其前提是對(duì)數(shù)據(jù)以及產(chǎn)生這些數(shù)據(jù)的具體業(yè)務(wù)場(chǎng)景有比較深入的理解。Spark中特征工程在ML中分為以下幾部分：從原始數(shù)據(jù)中提取特征從原特征中選擇最有效的一組子特征從已有的特征中構(gòu)建更有用的特征一種用于海量高維數(shù)據(jù)的近似最近鄰快速查找技術(shù)特征提取特征轉(zhuǎn)換特征選擇*局部敏感哈希6.4.1特征提取ML中常用的特征提取操作有以下4種：FeatureHasherWord2VecCountVectorizerTF-IDF特征提取TF-IDF是一種統(tǒng)計(jì)方法，用以評(píng)估一個(gè)字詞對(duì)于一個(gè)文件集或一個(gè)語(yǔ)料庫(kù)中的其中一份文件的重要程度。Word2Vec是一個(gè)基于神經(jīng)網(wǎng)絡(luò)的模型，用于從文本語(yǔ)料中學(xué)習(xí)單詞的向量表示。通過預(yù)測(cè)上下文單詞來學(xué)習(xí)每個(gè)單詞的向量，從而捕捉單詞之間的語(yǔ)義關(guān)系CountVectorizer是一種將文本數(shù)據(jù)轉(zhuǎn)換為詞頻矩陣的工具。它計(jì)算單詞在文檔中出現(xiàn)的次數(shù)，通常用于構(gòu)建詞袋模型。FeatureHasher，又稱為哈希特征，是一種使用哈希技術(shù)將特征轉(zhuǎn)換為固定長(zhǎng)度的向量的方法。適合于處理具有大量特征維度的數(shù)據(jù)6.4.2特征轉(zhuǎn)換ML中常用的24種特征轉(zhuǎn)換方法可以具體歸納為以下幾類操作：文檔轉(zhuǎn)換主要對(duì)文檔內(nèi)容數(shù)據(jù)進(jìn)行轉(zhuǎn)換。ML中主要文檔轉(zhuǎn)換的方法有Tokenizer、StopWordsRemover、n-gram、StringIndexer、IndexToString。正則化將每個(gè)樣本縮放到單位范數(shù)上，對(duì)于二次型（點(diǎn)積）或者其他核函數(shù)方法計(jì)算兩個(gè)樣本之間的相似性比較有用。ML中主要進(jìn)行正則化的方法有Normalizer。標(biāo)準(zhǔn)化是在不改變?cè)瓟?shù)據(jù)分布的前提下，將數(shù)據(jù)按比例縮放，使之落入一個(gè)限定的區(qū)間，使數(shù)據(jù)之間具有可比性。ML中主要使用標(biāo)準(zhǔn)化的方法有StandardScaler和RobustScaler。010203文檔轉(zhuǎn)換正則化標(biāo)準(zhǔn)化6.4.2特征轉(zhuǎn)換歸一化是對(duì)數(shù)據(jù)集進(jìn)行區(qū)間縮放，縮放到單位區(qū)間內(nèi)，把有單位的數(shù)據(jù)轉(zhuǎn)化為沒有單位的數(shù)據(jù)，即統(tǒng)一數(shù)據(jù)的衡量標(biāo)準(zhǔn)，消除單位的影響。ML中主要使用歸一化的方法有MinMaxScaler與MaxAbsScaler。離散化指的是將連續(xù)特征劃分離散的過程，將原始定量特征的一個(gè)區(qū)間一一映射到單一的值。ML中主要使用離散化的方法有QuantileDiscretizer、Bucketizer、Binarizer。獨(dú)熱編碼將分類轉(zhuǎn)換為二進(jìn)制向量，該向量中只有一位為1，其他為0。獨(dú)熱編碼一般在需要計(jì)算空間距離的算法中對(duì)原始的無先后順序的分類數(shù)據(jù)進(jìn)行處理。ML中主要使用獨(dú)熱編碼的方法有OneHotEncoderEstimator。040506歸一化離散化獨(dú)熱編碼6.4.2特征轉(zhuǎn)換缺失值補(bǔ)全用以處理數(shù)據(jù)集中存在的空缺或缺失值的問題。常用策略包括使用固定值填充、使用統(tǒng)計(jì)量（如平均數(shù)、中位數(shù)）填充或是通過預(yù)測(cè)模型估計(jì)缺失值。ML中主要使用缺失值補(bǔ)全的方法有Imputer。特征構(gòu)造通過對(duì)已有特征進(jìn)行拆分、組合構(gòu)建出新的特征。ML中主要使用特征構(gòu)造的方法有PolynomialExpansion、DiscreteCosineTransform(DCT)、Interaction、ElementwiseProduct、SQLTransformer、VectorAssembler。降維是指減少特征數(shù)量的同時(shí)盡量保留原數(shù)據(jù)的重要信息。ML中主要使用降維的方法有PCA。070809缺失值補(bǔ)全特征構(gòu)造降維6.4.3特征選擇VectorSlicer從特征向量中選擇特定子集，適用于已知重要特征的場(chǎng)景。1RFormula解析類似R語(yǔ)言的模型公式字符串，自動(dòng)處理分類特征，組合多個(gè)列為特征向量，適用于結(jié)構(gòu)化數(shù)據(jù)處理。2ChiSqSelector基于卡方檢驗(yàn)的特征選擇方法，通過卡方統(tǒng)計(jì)量選擇與目標(biāo)變量最相關(guān)的特征，適用于分類問題。3UnivariateFeature-Selector使用單變量統(tǒng)計(jì)測(cè)試（如方差分析F值、卡方檢驗(yàn)等）評(píng)估每個(gè)特征與目標(biāo)變量的關(guān)系，選擇最相關(guān)特征，適用于分類和回歸問題。4VarianceThreshold-Selector簡(jiǎn)單的特征選擇方法，通過設(shè)定方差閾值移除低方差特征，減少數(shù)據(jù)噪聲，適用于去除變化不大的特征。56.5常用機(jī)器學(xué)習(xí)算法ML提供了一系列常用機(jī)器學(xué)習(xí)算法，包括回歸、分類、聚類和協(xié)同過濾等，適用于各個(gè)領(lǐng)域的不同場(chǎng)景分類算法聚類算法回歸算法推薦算法6.5.1回歸算法回歸（Regression）算法是一種機(jī)器學(xué)習(xí)算法，用于根據(jù)輸入數(shù)據(jù)預(yù)測(cè)數(shù)值?；貧w算法試圖通過對(duì)數(shù)據(jù)擬合數(shù)學(xué)模型來找到輸入變量和輸出變量之間的關(guān)系?；貧w算法的目的是找到輸入特征和目標(biāo)變量之間的數(shù)學(xué)關(guān)系，可用于對(duì)新的數(shù)據(jù)進(jìn)行準(zhǔn)確預(yù)測(cè)。6.5.1回歸算法線性回歸線性回歸用于建立一個(gè)或多個(gè)自變量與因變量之間的線性關(guān)系。線性回歸模型的目的是找到一條直線（在二維空間中）或一個(gè)平面（在三維空間中），最好地?cái)M合數(shù)據(jù)點(diǎn)，用以預(yù)測(cè)連續(xù)型的目標(biāo)變量。決策樹回歸決策樹回歸是一種用于預(yù)測(cè)連續(xù)值輸出的機(jī)器學(xué)習(xí)方法。它與決策樹分類相似，但后者是用于預(yù)測(cè)離散類別標(biāo)簽。在決策樹回歸中，樹的葉子節(jié)點(diǎn)并不表示一個(gè)類別，而是代表了一個(gè)數(shù)值，通常是該節(jié)點(diǎn)下所有訓(xùn)練樣本目標(biāo)值的平均數(shù)。隨機(jī)森林回歸隨機(jī)森林回歸通過構(gòu)建多個(gè)決策樹并將它們的預(yù)測(cè)結(jié)果進(jìn)行平均來提高預(yù)測(cè)性能。在回歸任務(wù)中，隨機(jī)森林通過組合多棵決策樹的結(jié)果來預(yù)測(cè)連續(xù)數(shù)值。6.5.1回歸算法梯度提升樹回歸梯度提升樹回歸通過逐步添加新的模型來糾正之前模型的錯(cuò)誤，從而構(gòu)建一個(gè)預(yù)測(cè)性能更強(qiáng)的整體模型。生存回歸生存回歸，也稱為生存分析，用于分析直到某一事件發(fā)生的時(shí)間。適用于處理刪失數(shù)據(jù)，即那些在觀察結(jié)束時(shí)尚未發(fā)生目標(biāo)事件的數(shù)據(jù)點(diǎn)。單調(diào)回歸單調(diào)回歸是一種回歸分析方法，它用于在保持預(yù)測(cè)值的單調(diào)性的同時(shí)擬合數(shù)據(jù)。適用于那些已知或假設(shè)目標(biāo)變量與特征之間存在某種單調(diào)關(guān)系的情況。因子分解機(jī)回歸因子分解機(jī)回歸可以直接輸出連續(xù)型預(yù)測(cè)變量。其目標(biāo)是優(yōu)化回歸損失函數(shù)，如最小二乘誤差（MSE）。6.5.1回歸算法程序選用官方數(shù)據(jù)集（data/mllib/sample_linear_regression_data.txt)，每行數(shù)據(jù)表示一個(gè)訓(xùn)練樣本，其中第一列是目標(biāo)變量（標(biāo)簽），后面的列是特征，格式為數(shù)字:特征值。例如在第一行中，-9.490009878824548是目標(biāo)變量，表示樣本的標(biāo)簽。后面的數(shù)字（例如1:0.4551273600657362）表示特征索引和對(duì)應(yīng)的特征值。在這個(gè)例子中，有10個(gè)特征（索引1到10）6.5.1回歸算法importorg.apache.spark.sql.SparkSession//導(dǎo)入SparkSession，用于創(chuàng)建一個(gè)Spark應(yīng)用的入口點(diǎn)importorg.apache.spark.ml.regression.LinearRegression//導(dǎo)入LinearRegression，用于創(chuàng)建線性回歸模型objectLinearRegressionExample{defmain(args:Array[String]):Unit={valspark=SparkSession.builder.appName("LinearRegressionExample").master("local").getOrCreate()//加載訓(xùn)練數(shù)據(jù)valtraining=spark.read.format("libsvm")//指定數(shù)據(jù)格式為libsvm.load("data/mllib/sample_linear_regression_data.txt")//加載數(shù)據(jù)文件vallr=newLinearRegression()//創(chuàng)建線性回歸模型實(shí)例.setMaxIter(10)//設(shè)置最大迭代次數(shù)為10.setRegParam(0.3)//設(shè)置正則化參數(shù)為0.3.setElasticNetParam(0.8)//設(shè)置彈性網(wǎng)絡(luò)混合參數(shù)為0.8Scala6.5.1回歸算法//擬合模型vallrModel=lr.fit(training)//使用訓(xùn)練數(shù)據(jù)擬合模型//打印線性回歸模型的系數(shù)和截距println(s"Coefficients:${lrModel.coefficients}Intercept:${lrMercept}")//匯總模型在訓(xùn)練集上的統(tǒng)計(jì)信息，并打印一些指標(biāo)valtrainingSummary=lrModel.summary//獲取模型摘要println(s"numIterations:${trainingSummary.totalIterations}")//打印總迭代次數(shù)println(s"objectiveHistory:[${trainingSummary.objectiveHistory.mkString(",")}]")//打印目標(biāo)函數(shù)歷史記錄trainingSummary.residuals.show()//顯示模型殘差println(s"RMSE:${trainingSummary.rootMeanSquaredError}")//打印均方根誤差（RMSE）println(s"r2:${trainingSummary.r2}")//打印R平方值spark.stop()//停止SparkSession}}Scala6.5.1回歸算法frompyspark.sqlimportSparkSession#導(dǎo)入SparkSession類frompyspark.ml.regressionimportLinearRegression#導(dǎo)入LinearRegression類defmain():#定義主函數(shù)#創(chuàng)建SparkSession對(duì)象spark=SparkSession\.builder\.appName("LinearRegressionExample")\.master("local")\.getOrCreate()#讀取訓(xùn)練數(shù)據(jù)，使用libsvm格式training=spark.read.format("libsvm")\.load("data/mllib/sample_linear_regression_data.txt")#創(chuàng)建LinearRegression對(duì)象，設(shè)置最大迭代次數(shù)為10,正則化參數(shù)為0.3,彈性網(wǎng)絡(luò)混合參數(shù)為0.8lr=LinearRegression(maxIter=10,regParam=0.3,elasticNetParam=0.8)Python6.5.1回歸算法#使用訓(xùn)練數(shù)據(jù)擬合模型lrModel=lr.fit(training)#打印線性回歸模型的系數(shù)print("Coefficients:%s"%str(lrModel.coefficients))#打印線性回歸模型的截距print("Intercept:%s"%str(lrMercept))#獲取模型摘要對(duì)象trainingSummary=lrModel.summary#打印模型訓(xùn)練的迭代次數(shù)print("numIterations:%d"%trainingSummary.totalIterations)#打印目標(biāo)函數(shù)的歷史記錄print("objectiveHistory:%s"%str(trainingSummary.objectiveHistory))#顯示模型殘差trainingSummary.residuals.show()

Python6.5.1回歸算法#打印模型的均方根誤差（RMSE）print("RMSE:%f"%trainingSummary.rootMeanSquaredError)#打印模型的R平方值print("r2:%f"%trainingSummary.r2)#停止SparkSessionspark.stop()if__name__=="__main__":main()Python6.5.1回歸算法運(yùn)行結(jié)果：Coefficients：線性回歸模型中每個(gè)特征的系數(shù)。Intercept：線性回歸模型的截距，即當(dāng)所有特征都為0時(shí)的預(yù)測(cè)值。numIterations：模型訓(xùn)練時(shí)的迭代次數(shù)。objectiveHisto