Spark入門實戰系列--8.Spark MLlib（下）--機器學習庫SparkMLlib實戰

【注】該系列文章以及使用到安裝包/測試數據 能夠在《傾情大奉送--Spark入門實戰系列》獲取

1、MLlib實例

1.1 聚類實例

1.1.1 算法說明

聚類（Cluster analysis）有時也被翻譯爲簇類，其核心任務是：將一組目標object劃分爲若干個簇，每一個簇之間的object儘量類似，簇與簇之間的object儘量相異。聚類算法是機器學習（或者說是數據挖掘更合適）中重要的一部分，除了最爲簡單的K-Means聚類算法外，比較常見的還有層次法（CURE、CHAMELEON等）、網格算法（STING、WaveCluster等），等等。

較權威的聚類問題定義：所謂聚類問題，就是給定一個元素集合D，其中每一個元素具備n個可觀察屬性，使用某種算法將D劃分紅k個子集，要求每一個子集內部的元素之間相異度儘量低，而不一樣子集的元素相異度儘量高。其中每一個子集叫作一個簇。

K-means聚類屬於無監督學習，以往的迴歸、樸素貝葉斯、SVM等都是有類別標籤y的，也就是說樣例中已經給出了樣例的分類。而聚類的樣本中卻沒有給定y，只有特徵x，好比假設宇宙中的星星能夠表示成三維空間中的點集。聚類的目的是找到每一個樣本x潛在的類別y，並將同類別y的樣本x放在一塊兒。好比上面的星星，聚類後結果是一個個星團，星團裏面的點相互距離比較近，星團間的星星距離就比較遠了。

與分類不一樣，分類是示例式學習，要求分類前明確各個類別，並斷言每一個元素映射到一個類別。而聚類是觀察式學習，在聚類前能夠不知道類別甚至不給定類別數量，是無監督學習的一種。目前聚類普遍應用於統計學、生物學、數據庫技術和市場營銷等領域，相應的算法也很是多。

1.1.2 實例介紹

在該實例中將介紹K-Means算法，K-Means屬於基於平方偏差的迭代重分配聚類算法，其核心思想十分簡單：

l隨機選擇K箇中心點；

l計算全部點到這K箇中心點的距離，選擇距離最近的中心點爲其所在的簇；

l簡單地採用算術平均數（mean）來從新計算K個簇的中心；

l重複步驟2和3，直至簇類再也不發生變化或者達到最大迭代值；

l輸出結果。

K-Means算法的結果好壞依賴於對初始聚類中心的選擇，容易陷入局部最優解，對K值的選擇沒有準則可依循，對異常數據較爲敏感，只能處理數值屬性的數據，聚類結構可能不平衡。

本實例中進行以下步驟：

1.裝載數據，數據以文本文件方式進行存放；

2.將數據集聚類，設置2個類和20次迭代，進行模型訓練造成數據模型；

3.打印數據模型的中心點；

4.使用偏差平方之和來評估數據模型；

5.使用模型測試單點數據；

6.交叉評估1，返回結果；交叉評估2，返回數據集和結果。

1.1.3測試數聽說明

該實例使用的數據爲kmeans_data.txt，能夠在本系列附帶資源/data/class8/目錄中找到。在該文件中提供了6個點的空間位置座標，使用K-means聚類對這些點進行分類。

使用的kmeans_data.txt的數據以下所示：

0.0 0.0 0.0

0.1 0.1 0.1

0.2 0.2 0.2

9.0 9.0 9.0

9.1 9.1 9.1

9.2 9.2 9.2

1.1.4程序代碼

import org.apache.log4j.{Level, Logger}

import org.apache.spark.{SparkConf, SparkContext}

import org.apache.spark.mllib.clustering.KMeans

import org.apache.spark.mllib.linalg.Vectors

 

object Kmeans {

  def main(args: Array[String]) {

    // 屏蔽沒必要要的日誌顯示在終端上

    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)

    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

 

    // 設置運行環境

    val conf = new SparkConf().setAppName("Kmeans").setMaster("local[4]")

    val sc = new SparkContext(conf)

 

    // 裝載數據集

    val data = sc.textFile("/home/hadoop/upload/class8/kmeans_data.txt", 1)

    val parsedData = data.map(s => Vectors.dense(s.split(' ').map(_.toDouble)))

 

    // 將數據集聚類，2個類，20次迭代，進行模型訓練造成數據模型

    val numClusters = 2

    val numIterations = 20

    val model = KMeans.train(parsedData, numClusters, numIterations)

 

    // 打印數據模型的中心點

    println("Cluster centers:")

    for (c <- model.clusterCenters) {

      println("  " + c.toString)

    }

 

    // 使用偏差平方之和來評估數據模型

    val cost = model.computeCost(parsedData)

    println("Within Set Sum of Squared Errors = " + cost)

 

    // 使用模型測試單點數據

println("Vectors 0.2 0.2 0.2 is belongs to clusters:" + model.predict(Vectors.dense("0.2 0.2 0.2".split(' ').map(_.toDouble))))

println("Vectors 0.25 0.25 0.25 is belongs to clusters:" + model.predict(Vectors.dense("0.25 0.25 0.25".split(' ').map(_.toDouble))))

println("Vectors 8 8 8 is belongs to clusters:" + model.predict(Vectors.dense("8 8 8".split(' ').map(_.toDouble))))

 

    // 交叉評估1，只返回結果

    val testdata = data.map(s => Vectors.dense(s.split(' ').map(_.toDouble)))

    val result1 = model.predict(testdata)

   result1.saveAsTextFile("/home/hadoop/upload/class8/result_kmeans1")

 

    // 交叉評估2，返回數據集和結果

    val result2 = data.map {

      line =>

        val linevectore = Vectors.dense(line.split(' ').map(_.toDouble))

        val prediction = model.predict(linevectore)

        line + " " + prediction

    }.saveAsTextFile("/home/hadoop/upload/class8/result_kmeans2")

 

    sc.stop()

  }

}

1.1.5 IDEA執行狀況

第一步   使用以下命令啓動Spark集羣

$cd /app/hadoop/spark-1.1.0

$sbin/start-all.sh

第二步   在IDEA中設置運行環境

在IDEA運行配置中設置Kmeans運行配置，因爲讀入的數據已經在程序中指定，故在該設置界面中不須要設置輸入參數

第三步   執行並觀察輸出

在運行日誌窗口中能夠看到，經過計算計算出模型並找出兩個簇中心點：(9.1，9.1，9.1)和(0.1，0.1，0.1)，使用模型對測試點進行分類求出分屬於族簇。

第四步   查看輸出結果文件

在/home/hadoop/upload/class8目錄中有兩個輸出目錄：

查看結果1，在該目錄中只輸出告終果，分別列出了6個點所屬不一樣的族簇

查看結果2，在該目錄中輸出了數據集和結果

1.2 迴歸算法實例

1.2.1 算法說明

線性迴歸是利用稱爲線性迴歸方程的函數對一個或多個自變量和因變量之間關係進行建模的一種迴歸分析方法，只有一個自變量的狀況稱爲簡單迴歸，大於一個自變量狀況的叫作多元迴歸，在實際狀況中大多數都是多元迴歸。

線性迴歸（Linear Regression）問題屬於監督學習（Supervised Learning）範疇，又稱分類（Classification）或概括學習（Inductive Learning）。這類分析中訓練數據集中給出的數據類型是肯定的。機器學習的目標是，對於給定的一個訓練數據集，經過不斷的分析和學習產生一個聯繫屬性集合和類標集合的分類函數（Classification Function）或預測函數）Prediction Function），這個函數稱爲分類模型（Classification Model——或預測模型（Prediction Model）。經過學習獲得的模型能夠是一個決策樹、規格集、貝葉斯模型或一個超平面。經過這個模型能夠對輸入對象的特徵向量預測或對對象的類標進行分類。

迴歸問題中一般使用最小二乘（Least Squares）法來迭代最優的特徵中每一個屬性的比重，經過損失函數（Loss Function）或錯誤函數（Error Function)定義來設置收斂狀態，即做爲梯度降低算法的逼近參數因子。

1.2.2 實例介紹

該例子給出瞭如何導入訓練集數據，將其解析爲帶標籤點的RDD，而後使用了LinearRegressionWithSGD 算法來創建一個簡單的線性模型來預測標籤的值，最後計算了均方差來評估預測值與實際值的吻合度。

線性迴歸分析的整個過程能夠簡單描述爲以下三個步驟：

（1）尋找合適的預測函數，即上文中的 h(x) ，用來預測輸入數據的判斷結果。這個過程是很是關鍵的，須要對數據有必定的瞭解或分析，知道或者猜想預測函數的「大概」形式，好比是線性函數仍是非線性函數，如果非線性的則沒法用線性迴歸來得出高質量的結果。

（2）構造一個Loss函數（損失函數），該函數表示預測的輸出（h）與訓練數據標籤之間的誤差，能夠是兩者之間的差（h-y）或者是其餘的形式（如平方差開方）。綜合考慮全部訓練數據的「損失」，將Loss求和或者求平均，記爲 J(θ) 函數，表示全部訓練數據預測值與實際類別的誤差。

（3）顯然， J(θ) 函數的值越小表示預測函數越準確（即h函數越準確），因此這一步須要作的是找到 J(θ) 函數的最小值。找函數的最小值有不一樣的方法，Spark中採用的是梯度降低法（stochastic gradient descent，SGD)。

1.2.3程序代碼

import org.apache.log4j.{Level, Logger}

import org.apache.spark.{SparkContext, SparkConf}

import org.apache.spark.mllib.regression.LinearRegressionWithSGD

import org.apache.spark.mllib.regression.LabeledPoint

import org.apache.spark.mllib.linalg.Vectors

 

object LinearRegression {

  def main(args:Array[String]): Unit ={

    // 屏蔽沒必要要的日誌顯示終端上

    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)

    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

 

    // 設置運行環境

    val conf = new SparkConf().setAppName("Kmeans").setMaster("local[4]")

    val sc = new SparkContext(conf)

 

    // Load and parse the data

    val data = sc.textFile("/home/hadoop/upload/class8/lpsa.data")

    val parsedData = data.map { line =>

      val parts = line.split(',')

      LabeledPoint(parts(0).toDouble, Vectors.dense(parts(1).split(' ').map(_.toDouble)))

    }

 

    // Building the model

    val numIterations = 100

    val model = LinearRegressionWithSGD.train(parsedData, numIterations)

 

    // Evaluate model on training examples and compute training error

    val valuesAndPreds = parsedData.map { point =>

      val prediction = model.predict(point.features)

      (point.label, prediction)

    }

 

    val MSE = valuesAndPreds.map{ case(v, p) => math.pow((v - p), 2)}.reduce (_ + _) / valuesAndPreds.count

    println("training Mean Squared Error = " + MSE)

 

    sc.stop()

  }

}

1.2.4 執行狀況

第一步   啓動Spark集羣

$cd /app/hadoop/spark-1.1.0

$sbin/start-all.sh

第二步   在IDEA中設置運行環境

在IDEA運行配置中設置LinearRegression運行配置，因爲讀入的數據已經在程序中指定，故在該設置界面中不須要設置輸入參數

第三步   執行並觀察輸出

1.3 協同過濾實例

1.3.1 算法說明

協同過濾（Collaborative Filtering，簡稱CF，WIKI上的定義是：簡單來講是利用某個興趣相投、擁有共同經驗之羣體的喜愛來推薦感興趣的資訊給使用者，我的透過合做的機制給予資訊至關程度的迴應（如評分）並記錄下來以達到過濾的目的，進而幫助別人篩選資訊，迴應不必定侷限於特別感興趣的，特別不感興趣資訊的紀錄也至關重要。

協同過濾常被應用於推薦系統。這些技術旨在補充用戶—商品關聯矩陣中所缺失的部分。

MLlib 當前支持基於模型的協同過濾，其中用戶和商品經過一小組隱性因子進行表達，而且這些因子也用於預測缺失的元素。MLLib 使用交替最小二乘法（ALS） 來學習這些隱性因子。

用戶對物品或者信息的偏好，根據應用自己的不一樣，可能包括用戶對物品的評分、用戶查看物品的記錄、用戶的購買記錄等。其實這些用戶的偏好信息能夠分爲兩類：

l  顯式的用戶反饋：這類是用戶在網站上天然瀏覽或者使用網站之外，顯式地提供反饋信息，例如用戶對物品的評分或者對物品的評論。

l  隱式的用戶反饋：這類是用戶在使用網站是產生的數據，隱式地反映了用戶對物品的喜愛，例如用戶購買了某物品，用戶查看了某物品的信息，等等。

顯式的用戶反饋能準確地反映用戶對物品的真實喜愛，但須要用戶付出額外的代價；而隱式的用戶行爲，經過一些分析和處理，也能反映用戶的喜愛，只是數據不是很精確，有些行爲的分析存在較大的噪音。但只要選擇正確的行爲特徵，隱式的用戶反饋也能獲得很好的效果，只是行爲特徵的選擇可能在不一樣的應用中有很大的不一樣，例如在電子商務的網站上，購買行爲其實就是一個能很好表現用戶喜愛的隱式反饋。

推薦引擎根據不一樣的推薦機制可能用到數據源中的一部分，而後根據這些數據，分析出必定的規則或者直接對用戶對其餘物品的喜愛進行預測計算。這樣推薦引擎能夠在用戶進入時給他推薦他可能感興趣的物品。

MLlib目前支持基於協同過濾的模型，在這個模型裏，用戶和產品被一組能夠用來預測缺失項目的潛在因子來描述。特別是咱們實現交替最小二乘（ALS）算法來學習這些潛在的因子，在 MLlib 中的實現有以下參數：

l  numBlocks是用於並行化計算的分塊個數（設置爲-1時 爲自動配置）；

l  rank是模型中隱性因子的個數；

l  iterations是迭代的次數；

l  lambda是ALS 的正則化參數；

l  implicitPrefs決定了是用顯性反饋ALS 的版本仍是用隱性反饋數據集的版本；

l  alpha是一個針對於隱性反饋 ALS 版本的參數，這個參數決定了偏好行爲強度的基準。

1.3.2 實例介紹

在本實例中將使用協同過濾算法對GroupLens Research（http://grouplens.org/datasets/movielens/）提供的數據進行分析，該數據爲一組從20世紀90年底到21世紀初由MovieLens用戶提供的電影評分數據，這些數據中包括電影評分、電影元數據（風格類型和年代）以及關於用戶的人口統計學數據（年齡、郵編、性別和職業等）。根據不一樣需求該組織提供了不一樣大小的樣本數據，不一樣樣本信息中包含三種數據：評分、用戶信息和電影信息。

對這些數據分析進行以下步驟：

1. 裝載以下兩種數據：

a)裝載樣本評分數據，其中最後一列時間戳除10的餘數做爲key，Rating爲值；

b)裝載電影目錄對照表（電影ID->電影標題）

2.將樣本評分表以key值切分紅3個部分，分別用於訓練 (60%，並加入用戶評分), 校驗 (20%), and 測試 (20%)

3.訓練不一樣參數下的模型，並再校驗集中驗證，獲取最佳參數下的模型

4.用最佳模型預測測試集的評分，計算和實際評分之間的均方根偏差

5.根據用戶評分的數據，推薦前十部最感興趣的電影（注意要剔除用戶已經評分的電影）

1.3.3 測試數聽說明

在MovieLens提供的電影評分數據分爲三個表：評分、用戶信息和電影信息，在該系列提供的附屬數據提供大概6000位讀者和100萬個評分數據，具體位置爲/data/class8/movielens/data目錄下，對三個表數聽說明能夠參考該目錄下README文檔。

1.評分數聽說明（ratings.data)

該評分數據總共四個字段，格式爲UserID::MovieID::Rating::Timestamp，分爲爲用戶編號：：電影編號：：評分：：評分時間戳，其中各個字段說明以下：

l用戶編號範圍1~6040

l電影編號1~3952

l電影評分爲五星評分，範圍0~5

l評分時間戳單位秒

l每一個用戶至少有20個電影評分

使用的ratings.dat的數據樣本以下所示：

1::1193::5::978300760

1::661::3::978302109

1::914::3::978301968

1::3408::4::978300275

1::2355::5::978824291

1::1197::3::978302268

1::1287::5::978302039

1::2804::5::978300719

2.用戶信息(users.dat)

用戶信息五個字段，格式爲UserID::Gender::Age::Occupation::Zip-code，分爲爲用戶編號：：性別：：年齡：：職業::郵編，其中各個字段說明以下：

l用戶編號範圍1~6040

l性別，其中M爲男性，F爲女性

l不一樣的數字表明不一樣的年齡範圍，如：25表明25~34歲範圍

l職業信息，在測試數據中提供了21中職業分類

l地區郵編

使用的users.dat的數據樣本以下所示：

1::F::1::10::48067

2::M::56::16::70072

3::M::25::15::55117

4::M::45::7::02460

5::M::25::20::55455

6::F::50::9::55117

7::M::35::1::06810

8::M::25::12::11413

3.電影信息(movies.dat)

電影數據分爲三個字段，格式爲MovieID::Title::Genres，分爲爲電影編號：：電影名：：電影類別，其中各個字段說明以下：

l電影編號1~3952

l由IMDB提供電影名稱，其中包括電影上映年份

l電影分類，這裏使用實際分類名非編號，如：Action、Crime等

使用的movies.dat的數據樣本以下所示：

1::Toy Story (1995)::Animation|Children's|Comedy

2::Jumanji (1995)::Adventure|Children's|Fantasy

3::Grumpier Old Men (1995)::Comedy|Romance

4::Waiting to Exhale (1995)::Comedy|Drama

5::Father of the Bride Part II (1995)::Comedy

6::Heat (1995)::Action|Crime|Thriller

7::Sabrina (1995)::Comedy|Romance

8::Tom and Huck (1995)::Adventure|Children's

1.3.4 程序代碼

import java.io.File

import scala.io.Source

import org.apache.log4j.{Level, Logger}

import org.apache.spark.SparkConf

import org.apache.spark.SparkContext

import org.apache.spark.SparkContext._

import org.apache.spark.rdd._

import org.apache.spark.mllib.recommendation.{ALS, Rating, MatrixFactorizationModel}

 

object MovieLensALS {

 

  def main(args: Array[String]) {

    // 屏蔽沒必要要的日誌顯示在終端上

    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)

    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

 

    if (args.length != 2) {

      println("Usage: /path/to/spark/bin/spark-submit --driver-memory 2g --class week7.MovieLensALS " +

        "week7.jar movieLensHomeDir personalRatingsFile")

      sys.exit(1)

    }

 

    // 設置運行環境

    val conf = new SparkConf().setAppName("MovieLensALS").setMaster("local[4]")

    val sc = new SparkContext(conf)

 

    // 裝載用戶評分，該評分由評分器生成

    val myRatings = loadRatings(args(1))

    val myRatingsRDD = sc.parallelize(myRatings, 1)

 

    // 樣本數據目錄

    val movieLensHomeDir = args(0)

 

    // 裝載樣本評分數據，其中最後一列Timestamp取除10的餘數做爲key，Rating爲值,即(Int,Rating)

    val ratings = sc.textFile(new File(movieLensHomeDir, "ratings.dat").toString).map { line =>

      val fields = line.split("::")

      (fields(3).toLong % 10, Rating(fields(0).toInt, fields(1).toInt, fields(2).toDouble))

    }

 

    // 裝載電影目錄對照表（電影ID->電影標題）

    val movies = sc.textFile(new File(movieLensHomeDir, "movies.dat").toString).map { line =>

      val fields = line.split("::")

      (fields(0).toInt, fields(1))

    }.collect().toMap

 

    val numRatings = ratings.count()

    val numUsers = ratings.map(_._2.user).distinct().count()

    val numMovies = ratings.map(_._2.product).distinct().count()

 

    println("Got " + numRatings + " ratings from " + numUsers + " users on " + numMovies + " movies.")

 

    // 將樣本評分表以key值切分紅3個部分，分別用於訓練 (60%，並加入用戶評分), 校驗 (20%), and 測試 (20%)

    // 該數據在計算過程當中要屢次應用到，因此cache到內存

    val numPartitions = 4

    val training = ratings.filter(x => x._1 < 6)

      .values

      .union(myRatingsRDD) //注意ratings是(Int,Rating)，取value便可

      .repartition(numPartitions)

      .cache()

    val validation = ratings.filter(x => x._1 >= 6 && x._1 < 8)

      .values

      .repartition(numPartitions)

      .cache()

    val test = ratings.filter(x => x._1 >= 8).values.cache()

 

    val numTraining = training.count()

    val numValidation = validation.count()

    val numTest = test.count()

 

    println("Training: " + numTraining + ", validation: " + numValidation + ", test: " + numTest)

 

    // 訓練不一樣參數下的模型，並在校驗集中驗證，獲取最佳參數下的模型

    val ranks = List(8, 12)

    val lambdas = List(0.1, 10.0)

    val numIters = List(10, 20)

    var bestModel: Option[MatrixFactorizationModel] = None

    var bestValidationRmse = Double.MaxValue

    var bestRank = 0

    var bestLambda = -1.0

    var bestNumIter = -1

    for (rank <- ranks; lambda <- lambdas; numIter <- numIters) {

      val model = ALS.train(training, rank, numIter, lambda)

      val validationRmse = computeRmse(model, validation, numValidation)

      println("RMSE (validation) = " + validationRmse + " for the model trained with rank = "

        + rank + ", lambda = " + lambda + ", and numIter = " + numIter + ".")

      if (validationRmse < bestValidationRmse) {

        bestModel = Some(model)

        bestValidationRmse = validationRmse

        bestRank = rank

        bestLambda = lambda

        bestNumIter = numIter

      }

    }

 

    // 用最佳模型預測測試集的評分，並計算和實際評分之間的均方根偏差

    val testRmse = computeRmse(bestModel.get, test, numTest)

 

    println("The best model was trained with rank = " + bestRank + " and lambda = " + bestLambda  + ", and numIter = " + bestNumIter + ", and its RMSE on the test set is " + testRmse + ".")

 

    // create a naive baseline and compare it with the best model

    val meanRating = training.union(validation).map(_.rating).mean

    val baselineRmse =

      math.sqrt(test.map(x => (meanRating - x.rating) * (meanRating - x.rating)).mean)

    val improvement = (baselineRmse - testRmse) / baselineRmse * 100

    println("The best model improves the baseline by " + "%1.2f".format(improvement) + "%.")

 

    // 推薦前十部最感興趣的電影，注意要剔除用戶已經評分的電影

    val myRatedMovieIds = myRatings.map(_.product).toSet

    val candidates = sc.parallelize(movies.keys.filter(!myRatedMovieIds.contains(_)).toSeq)

    val recommendations = bestModel.get

      .predict(candidates.map((0, _)))

      .collect()

      .sortBy(-_.rating)

      .take(10)

 

    var i = 1

    println("Movies recommended for you:")

    recommendations.foreach { r =>

      println("%2d".format(i) + ": " + movies(r.product))

      i += 1

    }

 

  sc.stop()

  }

 

  /** 校驗集預測數據和實際數據之間的均方根偏差 **/

  def computeRmse(model: MatrixFactorizationModel, data: RDD[Rating], n: Long): Double = {

    val predictions: RDD[Rating] = model.predict(data.map(x => (x.user, x.product)))

    val predictionsAndRatings = predictions.map(x => ((x.user, x.product), x.rating))

      .join(data.map(x => ((x.user, x.product), x.rating)))

      .values

    math.sqrt(predictionsAndRatings.map(x => (x._1 - x._2) * (x._1 - x._2)).reduce(_ + _) / n)

  }

 

  /** 裝載用戶評分文件 **/

  def loadRatings(path: String): Seq[Rating] = {

    val lines = Source.fromFile(path).getLines()

    val ratings = lines.map { line =>

      val fields = line.split("::")

      Rating(fields(0).toInt, fields(1).toInt, fields(2).toDouble)

    }.filter(_.rating > 0.0)

    if (ratings.isEmpty) {

      sys.error("No ratings provided.")

    } else {

      ratings.toSeq

    }

  }

}

1.3.5 IDEA執行狀況

第一步   使用以下命令啓動Spark集羣

$cd /app/hadoop/spark-1.1.0

$sbin/start-all.sh

第二步   進行用戶評分，生成用戶樣本數據

因爲該程序中最終推薦給用戶十部電影，這須要用戶提供對樣本電影數據的評分，而後根據生成的最佳模型獲取當前用戶推薦電影。用戶可使用/home/hadoop/upload/class8/movielens/bin/rateMovies程序進行評分，最終生成personalRatings.txt文件：

第三步   在IDEA中設置運行環境

在IDEA運行配置中設置MovieLensALS運行配置，須要設置輸入數據所在文件夾和用戶的評分文件路徑：

l  輸入數據所在目錄：輸入數據文件目錄，在該目錄中包含了評分信息、用戶信息和電影信息，這裏設置爲/home/hadoop/upload/class8/movielens/data/

l  用戶的評分文件路徑：前一步驟中用戶對十部電影評分結果文件路徑，在這裏設置爲 /home/hadoop/upload/class8/movielens/personalRatings.txt

第四步   執行並觀察輸出

l  輸出Got 1000209 ratings from 6040 users on 3706 movies，表示本算法中計算數據包括大概100萬評分數據、6000多用戶和3706部電影；

l  輸出Training: 602252, validation: 198919, test: 199049，表示對評分數據進行拆分爲訓練數據、校驗數據和測試數據，大體佔比爲6:2:2；

l  在計算過程當中選擇8種不一樣模型對數據進行訓練，而後從中選擇最佳模型，其中最佳模型比基準模型提供22.30%

RMSE (validation) = 0.8680885498009973 for the model trained with rank = 8, lambda = 0.1, and numIter = 10.

RMSE (validation) = 0.868882967482595 for the model trained with rank = 8, lambda = 0.1, and numIter = 20.

RMSE (validation) = 3.7558695311242833 for the model trained with rank = 8, lambda = 10.0, and numIter = 10.

RMSE (validation) = 3.7558695311242833 for the model trained with rank = 8, lambda = 10.0, and numIter = 20.

RMSE (validation) = 0.8663942501841964 for the model trained with rank = 12, lambda = 0.1, and numIter = 10.

RMSE (validation) = 0.8674684744165418 for the model trained with rank = 12, lambda = 0.1, and numIter = 20.

RMSE (validation) = 3.7558695311242833 for the model trained with rank = 12, lambda = 10.0, and numIter = 10.

RMSE (validation) = 3.7558695311242833 for the model trained with rank = 12, lambda = 10.0, and numIter = 20.

The best model was trained with rank = 12 and lambda = 0.1, and numIter = 10, and its RMSE on the test set is 0.8652326018300565.

The best model improves the baseline by 22.30%.

l  利用前面獲取的最佳模型，結合用戶提供的樣本數據，最終推薦給用戶以下影片：

Movies recommended for you:

 1: Bewegte Mann, Der (1994)

 2: Chushingura (1962)

 3: Love Serenade (1996)

 4: For All Mankind (1989)

 5: Vie est belle, La (Life is Rosey) (1987)

 6: Bandits (1997)

 7: King of Masks, The (Bian Lian) (1996)

 8: I'm the One That I Want (2000)

 9: Big Trees, The (1952)

10: First Love, Last Rites (1997)

 

2、參考資料

（1）Spark官網 mlllib說明  http://spark.apache.org/docs/1.1.0/mllib-guide.html

（2）《機器學習常見算法分類彙總》 http://www.ctocio.com/hotnews/15919.html