機器學習實踐：Spark MLlib庫介紹與使用
1、實驗描述
- MLlib （ Machine Learning Library ）是 Spark 的一個機器學習庫。它能夠較容易地解決一些實際的大規模機器學習問題。本實驗旨在學習 Spark 的機器學習庫—— MLlib 的相關知識，了解 MLlib 與 ML 之間的差別和聯系，掌握 MLlib 中的幾個基本資料類型
- 實驗時長：90分鐘
- 主要步驟：
- 學習Mllib的基本資料類型
- 學習Mllib的基本算法庫
- 利用Mllib算法庫中的協同推薦算法為使用者推薦電影
2、實驗環境
- 虛拟機數量：1
- 系統版本：CentOS 7.5
- Spark版本：Apache Spark 2.1.1
3、相關技能
- linux常用指令
- Spark shell
- Scala語言程式設計
- MLlib 庫的使用
- 協同過濾算法
- Spark SQL應用
4、相關知識點
- 資料類型：本地向量、标簽點、本地矩陣
- 稠密向量、稀疏向量
- MLlib 庫的主要内容
- MLlib 的資料類型和算法種類
- 協同過濾算法
- Spark SQL
5、效果圖
- 利用ALS模型對測試資料做預測，看到這個誤內插補點是在可以接受的範圍内的。詳情請看實驗步驟
圖 1
6、實驗步驟
6.1MLlib 的資料類型
6.1.1本地向量
6.1.1.1本地向量（Local Vector）的索引是整型的、從0開始的。而它的值為 Double 類型，存儲于單個機器内。 MLlib 支援兩種本地向量：稠密向量和稀疏向量
6.1.1.2本地向量的基類是 Vector 類，DenseVector 和 SparseVector 類是其兩種不同的實作。官方文檔推薦大家使用 Vector 類中已實作的工廠方法來建立本地向量
6.1.1.3安裝spark（這裡我們隻需要單機模式的spark的環境即可），解壓tgz下的spark安裝包
[[email protected] ~]$ cd tgz/spark
[[email protected] spark]$ tar -zxvf spark-2.1.1-bin-hadoop-2.7 -C ~/
[[email protected] spark]$ cd
6.1.1.4打開spark shell終端
[[email protected] ~]$ cd spark-2.1.1-bin-hadoop2.7/
[[email protected] spark-2.1.1-bin-hadoop2.7]$ bin/spark-shell
6.1.1.5進入spark shell指令行後，導入spark mllib中的本地向量相關包
6.1.1.6利用dense方法建立稠密向量(1.0, 0.0, 3.0)
6.1.1.7通過指定非零實體對應的索引和值，來建立稀疏向量 (1.0, 0.0, 3.0)
6.1.1.8通過指定非零實體，來建立稀疏向量 (1.0, 0.0, 3.0)
圖 2
6.1.1.9向量 (1.0, 0.0, 3.0) 寫成稠密形式就是 [1.0, 0.0, 3.0]，而寫成稀疏形式則是(3, [0, 2], [1.0, 3.0])，後者的第一個 3 是指向量的大小。稀疏和稠密的界限沒有嚴格意義上的标準，通常需要依據具體的問題來決定
6.1.1.10在Spark中，如果沒有引入org.apache.spark.mllib.linalg.Vector 包的話，Vector 會被認為是scala.collection.immutable.Vector 中定義的那個。是以，應當引入前者以顯式地使用 MLlib 中的向量
6.1.2标簽點
6.1.2.1标簽點（Labeled Point）是一個本地向量，也分稀疏或者稠密，并且是一個帶有标簽的本地向量。
6.1.2.2在 MLlib 中，标簽點常用于監督學習類算法。标簽（Label）是用 Double 類型存放的，是以标簽點可以用于回歸或者分類算法中。如果是二維分類，标簽則必須是 0 或 1 之間的一種。而如果是多個次元的分類，标簽應當是從 0 開始的數字，代表各個分類的索引。
6.1.2.3标簽點是由一個名為 LabeledPoint的 Case Class 樣例類定義的
6.1.2.4在spark-shell中建立标簽點， 引入标簽點相關的包
scala>import org.apache.spark.mllib.linalg.Vectors
scala>import org.apache.spark.mllib.regression.LabeledPoint
6.1.2.5建立一個帶有正面标簽和稠密特征向量的标簽點
6.1.2.6建立一個帶有負面标簽和稀疏特征向量的标簽點
圖 3
6.1.3本地矩陣
6.1.3.1本地矩陣（Local matrix）的索引也是從0開始的，并且是整型。值也為 Double 類型，存儲于單個機器内。 Mllib 支援兩種本地矩陣：稠密矩陣和稀疏矩陣
6.1.3.2稠密矩陣的實體值以列為主序的形式，存放于單個 Double 型數組内。稀疏矩陣的非零實體以列為主序的形式，存放于壓縮稀疏列（Compressed Sparse Column, CSC）中
6.1.3.3本地矩陣的基類是 Matrix 類，在 Spark 中有其兩種實作，分别是 DenseMatrix 和 SparseMatrix 。官方文檔中推薦使用已在 Matrices 類中實作的工廠方法來建立本地矩陣。需要注意的是，MLlib 中的本地矩陣是列主序的（column-major）
6.1.3.4在spark shell中建立本地矩陣，首先導入矩陣相關包
6.1.3.5建立稠密矩陣((1.0, 2.0), (3.0, 4.0), (5.0, 6.0))
6.1.3.6建立稀疏矩陣 ((9.0, 0.0), (0.0, 8.0), (0.0, 6.0))
圖 4
6.2MLlib算法
6.2.1MLlib 的算法涵蓋了基本統計、分類和回歸和協同過濾等 9 個大類，而更加新穎和高效的算法也在不斷地推陳出新。在 Spark中，主要的算法有：
6.2.1.1基本統計算法
- 彙總統計
- 相關統計
- 分層抽樣
- 假設檢驗
- 随機資料生成
- 核密度估計
6.2.1.2分類和回歸
6.2.1.3協同過濾
6.2.1.4聚類
6.2.1.5降維
6.2.1.6特征提取和轉化
6.2.1.7頻繁項挖掘
6.2.1.8評價名額
6.2.1.9PMML導出
6.2.2由于篇幅有限且算法内容較複雜，這裡就不一一講解，本實驗介紹使用一種交替最小二乘法的協同過濾算法
6.3執行個體：利用算法來推薦電影
6.3.1利用**:quit**指令退出spark shell指令行，解壓experiment/mllib目錄下的資料集檔案
scala>:quit
[[email protected] spark-2.1.1-bin-hadoop2.7]$ cd ~/experiment/mllib
[[email protected] mllib]$ unzip ml-1m.zip
[[email protected] mllib]$ cd ml-1m
6.3.2解壓完成之後有三個dat檔案，分别是movies.dat、ratings.data、users.dat
6.3.3我們檢視下使用者資訊users.dat檔案的前10條，每行的資料格式為`使用者ID：：性别：：年齡：：工作：：郵編`
[[email protected] ml-1m]$ head -10 users.dat
圖 5
6.3.3.1性别部分：M代表男性，F代表女性；
6.3.3.2年齡部分：1表示18歲以下，18表示18-24歲，25表示25-34歲，35表示35-44歲，45表示45-49歲，50表示50-55歲，56表示56歲及以上
6.3.3.3職業部分：職業編号分布在0-20，每一種編号分别表示一種職業，若想檢視詳情，打開README檔案，找到對users.dat的描述的内容
6.3.4我們檢視下評論資訊ratings.dat檔案的前10條，其格式為`使用者ID：：電影ID：：評論星級：：時間戳`
[[email protected] ml-1m]$ head -10 ratings.dat
圖 6
6.3.4.1電影ID部分：電影ID的範圍在1-3952之間，每一個ID代表一部電影
6.3.4.2評論星級部分：共五星用1-5的數字表示
6.3.4.3時間戳部分：評論釋出的時間戳
6.3.5我們檢視下電影資訊movies.dat檔案的前10條，其格式為`電影ID：：電影标題：：電影流派`
[[email protected] ml-1m]$ head -10 movies.dat
圖 7
6.3.5.1電影标題部分：電影的釋出名稱和釋出年份等
6.3.5.2電影流派部分：共18個流派，一部電影可能屬于一個或者多個流派，各流派如下：Action（行動）、Adventure（冒險）、Animation（動畫）、Children’s（兒童）、Comedy（喜劇）、Crime（犯罪）、Documentary（紀錄片）、Drama（戲劇）、Fantasy（幻想）、Film-Noir（黑色電影）、Horror（恐怖）、Musical（音樂）、Mystery（神秘）、Romance（浪漫）、Sci-Fi（科幻）、Thriller（驚悚）、War（戰争）、Western（西方）
6.3.6重新進入Spark shell
[[email protected] ml-1m]$ cd ~/spark-2.1.1-bin-hadoop2.7/
[[email protected] spark-2.1.1-bin-hadoop2.7]$ bin/spark-shell
6.3.7開始編寫實驗代碼
6.3.7.1引入本節實驗要使用的相關spark包
scala>import spark.implicits._
scala>import org.apache.spark.rdd._
scala>import org.apache.spark.sql._
scala>import org.apache.spark.mllib.recommendation.Rating
scala>import org.apache.spark.mllib.recommendation.ALS
scala>import org.apache.spark.mllib.recommendation.MatrixFactorizationModel
6.3.7.2首先對電影資料建立一個 Case Class，它對應了 movies.dat 檔案中的部分字段（其中 Genres（流派）字段不是我們所需要的，是以在 Case Class 中沒有設定對應的成員變量）
圖 8
6.3.7.3同樣的，我們為使用者資料建立一個 Case Class，它對應了 users.dat 檔案中的部分字段，（本實驗隻使用了 使用者ID 作為運算時的輸入）
圖 9
6.3.7.4我們不需定義ratings.dat資料的Case Class，因為評價類型的資料在spark機器學習架構中已經有org.apache.spark.mllib.recommendation.Rating類來做映射，在一開始的導包中我們已經導入了該包
6.3.7.5定義三個解析資料函數。這些函數将用于解析資料集中的每一行内容，去掉分隔符 ::，然後把資料映射到 Case Class 的對應成員中，是以，針對三種資料，分别編寫對應解析函數
// 解析 movies.dat 檔案的函數；參數為此檔案中的一行内容
scala> def parseMovieData(data: String): Movie = {
val dataField = data.split("::")
assert(dataField.size == 3)
Movie(dataField(0).toInt, dataField(1))
}
// 解析 users.dat 檔案的函數；參數為此檔案中的一行内容
scala> def parseUserData(data: String): User = {
val dataField = data.split("::")
assert(dataField.size == 5)
User(dataField(0).toInt, dataField(1), dataField(2).toInt, dataField(3).toInt, dataField(4))
}
// 解析 ratings.dat 檔案的函數，這裡用到的 Rating 類是在org.apache.spark.mllib.recommendation.Rating 包中定義的
scala> def parseRatingData(data: String): Rating = {
val dataField = data.split("::")
Rating(dataField(0).toInt, dataField(1).toInt, dataField(2).toDouble)
}
圖 10
6.3.8定義好三種資料對應的解析函數後，下面我們将他們導入到 RDD 中
scala>val moviesData = sc.textFile("/home/zkpk/experiment/mllib/ml-1m/movies.dat").map(parseMovieData).cache() //cache()将RDD緩存起來
scala>val usersData = sc.textFile("/home/zkpk/experiment/mllib/ml-1m/users.dat").map(parseUserData).cache()
scala>val ratingsData = sc.textFile("/home/zkpk/experiment/mllib/ml-1m/ratings.dat").map(parseRatingData).cache()
圖 11
6.3.9通過一些方法來檢視資料的某些特征
6.3.9.1看一下評價資料到底有多少個
圖 12
6.3.9.2看一下多有少個使用者評價了電影
圖 13
6.3.10将這些 RDD 轉化為 DataFrame ，用于後續的操作
scala>val moviesDF = moviesData.toDF()
scala>val usersDF = usersData.toDF()
scala>val ratingsDF = ratingsData.toDF()
圖 14
6.3.11将這幾個 DataFrame 注冊為臨時表
scala>moviesDF.registerTempTable("movies")
scala>usersDF.registerTempTable("users")
scala>ratingsDF.registerTempTable("ratings")
圖 15
6.3.12利用一個 SQL 查詢，擷取id是1680的使用者的評價高于 4.5 分的電影有哪些
6.3.13檢索完成後，調用 show() 函數檢視前20條結果，結果格式為：使用者ID，電影ID，使用者評分，電影标題（每次實驗結果可能略有不同）
圖 16
6.3.14開始機器學習，将已有的資料劃分為兩部分：訓練集、測試集
6.3.15繼續在spark shell中程式設計，将評分資料集按照訓練集70%，測試集30%來随機切分，250L是随機種子數
6.3.15.1這裡用到了 randomSplit 函數，兩個參數分别為劃分比例、産生随機數的種子值
6.3.16然後我們将劃分的結果存放在兩個不同的變量中：
scala>val trainingSetOfRatingsData = tempPartitions(0).cache() // 訓練集
scala>val testSetOfRatingData = tempPartitions(1).cache() // 測試集
6.3.17将訓練集用于 ALS 算法
6.3.17.1MLlib 中的 ALS 算法需要設定三個參數，分别是特征矩陣的秩數量 setRank、疊代次數 setIterations 和訓練用的訓練資料集。其中特征矩陣的秩設定為 20，表明生成的 userFeatures 和 itemFeatures 的個數是20
6.3.18生成推薦結果，通過 MatrixFactorizationModel 類的 recommendProducts 方法，我們可以對某個使用者生成推薦結果（每次實驗結果可能略有不同）
圖 17
6.3.18.11234 是我們随意指定的一個使用者 ID，做實驗時也可以修改成其他使用者的 ID 。10 是産生推薦的數量
6.3.19也可以利用mkString展現更直覺的推薦結果
圖 18
6.3.19.1輸出資料的格式是 Rating 類的格式：使用者ID、電影ID和該使用者對電影可能的評分。
6.3.20但僅看到電影的 ID，并不知道具體是什麼電影。是以還需要按照電影的 ID找到對應的電影标題:
scala>val movieTitles = moviesDF.rdd.map(array => (array(0), array(1))).collectAsMap() // 0、1分别是movieId、title
scala>val recomResultWithTitle = recomResult.map(rating => (movieTitles(rating.product), rating.rating))
6.3.21然後再把結果輸出，就可以看到推薦的電影名稱了：
圖 19
6.3.22模型評價，評價的方式便是将上邊得到的推薦結果（評分預測值）與測試集中實際結果相比較。利用 MatrixFactorizationModel 類的 predict 函數來得到預測的評分值
6.3.23得到對測試集的預測評分結果後，用 map 、 join算子将它與測試集的原始資料組合成 **((使用者ID, 電影ID), (測試集原有評分, 預測評分))**的資料結構。這個格式是 Key-Value 形式的，Key 為 (user, product)。我們是要把這裡的測試集原有評分與預測時得到的評分相比較，二者的聯系就是 user 和 product 相同。第一個 map 可以将 (使用者ID, 電影ID) 與 測試集原有評分 組合成 KV 格式；第二個 map 操作，可以将 (使用者ID, 電影ID) 與 預測評分 組合成 KV 格式
// 測試集
scala>val formatResultOfTestSet = testSetOfRatingData.map {
case Rating(user, product, rating) => ((user, product), rating)
}
// 預測結果
scala>val formatResultOfPredictionResult = predictResultOfTestSet.map {
case Rating(user, product, rating) => ((user, product), rating)
}
6.3.24然後利用join算子将二者連接配接起來，Key 相同的不同值就整合在一起形成了 ((使用者ID, 電影ID), (測試集原有評分, 預測評分)) 的格式。
6.3.25最後利用這個結果來計算預測評分和實際評分之間的平均絕對誤差。平均絕對誤差（Mean Absolute Error）是所有單個觀測值與算術平均值偏差的絕對值的平均。與平均誤差相比，平均絕對誤差由于被絕對值化，不會出現正負相抵消的情況，是以平均絕對誤差能更好地反映預測值誤差的實際情況。我們直接取finalResultForComparison 結果中 ratingOfTest 和 ratingOfPrediction兩個值，先算誤差，再取絕對值。最後對所有的絕對值計算平均數
scala>val MAE = finalResultForComparison.map {
case ((user, product), (ratingOfTest, ratingOfPrediction)) => {val deviation= (ratingOfTest - ratingOfPrediction)
Math.abs(deviation)}
}.mean()
6.3.26檢視終端結果（每次實驗結果可能略有不同），可以看到這個誤內插補點是在可以接受的範圍内的
圖 20
7、總結
算術平均值偏差的絕對值的平均。與平均誤差相比，平均絕對誤差由于被絕對值化，不會出現正負相抵消的情況，是以平均絕對誤差能更好地反映預測值誤差的實際情況。我們直接取finalResultForComparison 結果中 ratingOfTest 和 ratingOfPrediction兩個值，先算誤差，再取絕對值。最後對所有的絕對值計算平均數