基于Spark的电影推荐系统实现
基于Spark的電影推薦系統實現
- 一、業務場景
- 二、數據集說明
- 三、操作步驟
- 階段一、啟動HDFS、Spark集群服務和zeppelin服務器
- 階段二、準備案例中用到的數據集
- 階段三、對數據集進行探索和分析
未經許可，禁止以任何形式轉載，若要引用，請標注鏈接地址
全文共計7821字，閱讀大概需要3分鐘
一、業務場景
受某視頻網站委托，需開發一套基于Spark的大數據機器學習系統，應用協同過濾算法對網站客戶進行電影推薦。
二、數據集說明
本案例所使用的數據集說明如下：
??評分數據集：/data/dataset/ml/movielens/ratings.csv
??電影數據集：/data/dataset/ml/movielens/movies.csv
三、操作步驟
階段一、啟動HDFS、Spark集群服務和zeppelin服務器
1、啟動HDFS集群
??在Linux終端窗口下，輸入以下命令，啟動HDFS集群：
2、啟動Spark集群
??在Linux終端窗口下，輸入以下命令，啟動Spark集群：
3、啟動zeppelin服務器
??在Linux終端窗口下，輸入以下命令，啟動zeppelin服務器：
4、驗證以上進程是否已啟動
??在Linux終端窗口下，輸入以下命令，查看啟動的服務進程：
如果顯示以下6個進程，則說明各項服務啟動正常，可以繼續下一階段。
1. 2288 NameNode 2. 2402 DataNode 3. 2603 SecondaryNameNode 4. 2769 Master 5. 2891 Worker 6. 2984 ZeppelinServer階段二、準備案例中用到的數據集
1、將本案例要用到的數據集上傳到HDFS文件系統的”/data/dataset/ml/“目錄下。在Linux終端窗口下，輸入以下命令：
1. $ hdfs dfs -mkdir -p /data/dataset/ml 2. $ hdfs dfs -put /data/dataset/ml/movielens /data/dataset/ml/2、在Linux終端窗口下，輸入以下命令，查看HDFS上是否已經上傳了該數據集：
1. $ hdfs dfs -ls /data/dataset/ml/movielens這時應該看到movielens目錄及其中的訓練數據集已經上傳到了HDFS的”/data/dataset/ml/“目錄下。
階段三、對數據集進行探索和分析
1、新建一個zeppelin notebook文件，并命名為movie_project。
??2、先導入案例中要用到的機器學習庫。在notebook單元格中，輸入以下代碼：
同時按下【Shift+Enter】鍵，執行以上代碼，輸出內容如下：
import org.apache.spark.mllib.evaluation.RankingMetrics import org.apache.spark.ml.evaluation.RegressionEvaluator import org.apache.spark.ml.recommendation.ALS import org.apache.spark.ml.tuning.{ParamGridBuilder, CrossValidator} import org.apache.spark.sql.functions. import org.apache.spark.sql.types. import spark.implicits._3、加載數據集。在notebook單元格中，輸入以下代碼：
1. // 加載數據。因為不需要timestamp列，因此立即刪除它 2. val file = "hdfs://localhost:9000/data/dataset/ml/movielens/ratings.csv" 3. 4. val ratingsDF1 = spark.read.option("header", "true"). 5. option("inferSchema", "true"). 6. csv(file). 7. drop("timestamp") 8. 9. ratingsDF1.count 10. ratingsDF1.printSchema同時按下【Shift+Enter】鍵，執行以上代碼，輸出內容如下：
file: String = /data/spark_demo/movielens/ratings.csv ratingsDF1: org.apache.spark.sql.DataFrame = [userId: int, movieId: int … 1 more field] res97: Long = 100836 root|— userId: integer (nullable = true)|— movieId: integer (nullable = true)|— rating: double (nullable = true)由以上輸出內容可以看出，該數據集共有3個字段，分別是用戶id、電影id和該用戶對該電影的評分。
4、查看前5條數據。在notebook單元格中，輸入以下代碼：
1. ratingsDF1.show(5)同時按下【Shift+Enter】鍵，執行以上代碼，輸出內容如下：
5、對數據進行簡單統計。在notebook單元格中，輸入以下代碼：
1. // 看看被評分的電影總共有多少部： 2. ratingsDF1.select("movieId").distinct().count 3. 4. // 看看有多少用戶參與評分： 5. ratingsDF1.select("userId").distinct().count同時按下Shift+Enter鍵，執行以上代碼，輸出內容如下：
Long = 9724 Long = 610由以上輸出內容可以看出，該評分數據集中，參與的用戶有610名，被評論的電影有9724部。
6、快速檢查誰是活躍的電影評分者。在notebook單元格中，輸入以下代碼：
1. // 快速檢查誰是活躍的電影評分者 2. val ratingsByUserDF = ratingsDF1.groupBy("userId").count() 3. ratingsByUserDF.orderBy($"count".desc).show(10)同時按下【Shift+Enter】鍵，執行以上代碼，輸出內容如下：
由以上輸出內容可以看出，參與度最高用的用戶其id是414。
7、分析每部電影的最大評分數量。在notebook單元格中，輸入以下代碼：
1. // 分析每部電影的最大評分數量 2. val ratingsByMovieDF = ratingsDF1.groupBy("movieId").count() 3. ratingsByMovieDF.orderBy($"count".desc).show(10)同時按下【Shift+Enter】鍵，執行以上代碼，輸出內容如下：
由以上輸出內容可以看出，評價數超過300的電影有三部，其中評價數最多的電影其id是356。
8、數據拆分，將原始數據集拆分為訓練集和測試集，其中訓練集占80%，測試集占20%。在notebook單元格中，輸入以下代碼：
1. // 為訓練和測試準備數據 2. val Array(trainingData, testData) = ratingsDF1.randomSplit(Array(0.8, 0.2))同時按下【Shift+Enter】鍵，執行以上代碼，輸出內容如下：
trainingData: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [userId: int, movieId: int … 1 more field] testData: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [userId: int, movieId: int … 1 more field]9、Spark實現了ALS算法(Alternating Least Square)，這一步建立ALS的實例。在notebook單元格中，輸入以下代碼：
1. // 建立ALS的實例 2. val als = new ALS().setRank(12). 3. setMaxIter(10). 4. setRegParam(0.03). 5. setUserCol("userId"). 6. setItemCol("movieId"). 7. setRatingCol("rating")同時按下【Shift+Enter】鍵，執行以上代碼，輸出內容如下：
als: org.apache.spark.ml.recommendation.ALS = als_10a2c5c69e4010、訓練模型，并設置模型的冷啟動策略。在notebook單元格中，輸入以下代碼：
1. // 訓練模型 2. val model = als.fit(trainingData) 3. 4. // 從Spark 2.2.0開始，可以將coldStartStrategy參數設置為drop，以便刪除包含NaN值的預測的DataFrame中的任何行。 5. // 然后將在非nan數據上計算評估度量，該度量將是有效的。 6. model.setColdStartStrategy("drop")同時按下【Shift+Enter】鍵，執行以上代碼，輸出內容如下：
model: org.apache.spark.ml.recommendation.ALSModel = als_10a2c5c69e40 res169: model.type = als_10a2c5c69e4011、執行預測，并查看預測結果。在notebook單元格中，輸入以下代碼：
1. // 執行預測 2. val predictions = model.transform(testData) 3. 4. // 查看預測結果 5. predictions.sort("userId").show(10)同時按下【Shift+Enter】鍵，執行以上代碼，輸出內容如下：
由以上輸出內容可以看出，rating列為標簽列，prediction為預測結果列。
12、有的預測值為NaN(非數字)，這會影響到rmse的計算，因些需要先刪除結果集中的NaN值。在notebook單元格中，輸入以下代碼：
1. val predictions_dropNaN = predictions.na.drop(Array("prediction")) 2. predictions_dropNaN.count同時按下【Shift+Enter】鍵，執行以上代碼，輸出內容如下：
predictions_dropNaN: org.apache.spark.sql.DataFrame = [userId: int, movieId: int … 2 more fields] res178: Long = 19333由以上輸出內容可以看出，刪除prediction列具有NaN值的記錄以后，結果集中還有19333條記錄。
13、設置一個評估器（evaluator）來計算RMSE度量指標。在notebook單元格中，輸入以下代碼：
1. // 設置一個評估器（evaluator）來計算RMSE度量指標 2. val evaluator = new RegressionEvaluator().setMetricName("rmse"). 3. setLabelCol("rating"). 4. setPredictionCol("prediction") 5. 6. val rmse = evaluator.evaluate(predictions_dropNaN) 7. println(s"Root-mean-square error = ${rmse}")同時按下【Shift+Enter】鍵，執行以上代碼，輸出內容如下：
evaluator: org.apache.spark.ml.evaluation.RegressionEvaluator = regEval_7943cc497104 rmse: Double = 1.017470307395966 Root-mean-square error = 1.017470307395966由以上輸出內容可以看出，根均方差(rmse)值為
1.01747030739596614、使用ALSModel來執行推薦。在notebook單元格中，輸入以下代碼：
1. // 為所有用戶推薦排名前五的電影 2. model.recommendForAllUsers(5).show(false)同時按下【Shift+Enter】鍵，執行以上代碼，輸出內容如下：
15、為每部電影推薦top 3個用戶。在notebook單元格中，輸入以下代碼：
1. // 為每部電影推薦top 3個用戶 2. val recMovies = model.recommendForAllItems(3) 3. recMovies.show(5,false)同時按下【Shift+Enter】鍵，執行以上代碼，輸出內容如下：
由以上輸出內容可以看出，為每部電影推薦前3個用戶。但是不好的一點是，我們只能看到電影的id，而不是電影的名稱。
16、讀取電影數據集，這樣就能看到電影標題了。在notebook單元格中，輸入以下代碼：
1. // 讀取電影數據集，這樣就能看到電影標題了 2. val moviesFile = "hdfs://localhost:9000/data/dataset/ml/movielens/movies.csv" 3. val moviesDF = spark.read.option("header", "true").option("inferSchema", "true").csv(moviesFile) 4. 5. val recMoviesWithInfoDF = recMovies.join(moviesDF, "movieId") 6. recMoviesWithInfoDF.select("movieId", "title", "recommendations").show(5, false)同時按下【Shift+Enter】鍵，執行以上代碼，輸出內容如下：
由以上輸出內容可以看出，現在每部電影的id和名稱都是可見的了。
17、使用CrossValidator對ALS模型進行調優。在notebook單元格中，輸入以下代碼：
1. // 使用CrossValidator對ALS模型進行調優 2. val paramGrid = new ParamGridBuilder() 3. .addGrid(als.regParam, Array(0.05, 0.15)) 4. .addGrid(als.rank, Array(12,20)) 5. .build 6. 7. val crossValidator=new CrossValidator() 8. .setEstimator(als) 9. .setEvaluator(evaluator) 10. .setEstimatorParamMaps(paramGrid) 11. .setNumFolds(3) 12. 13. // 打印出4個超參數組合 14. crossValidator.getEstimatorParamMaps.foreach(println)同時按下【Shift+Enter】鍵，執行以上代碼，輸出內容如下：
paramGrid: Array[org.apache.spark.ml.param.ParamMap] = Array({als_10a2c5c69e40-rank: 12,als_10a2c5c69e40-regParam: 0.05 }, {als_10a2c5c69e40-rank: 20,als_10a2c5c69e40-regParam: 0.05 }, {als_10a2c5c69e40-rank: 12,als_10a2c5c69e40-regParam: 0.15 }, {als_10a2c5c69e40-rank: 20,als_10a2c5c69e40-regParam: 0.15 }) crossValidator: org.apache.spark.ml.tuning.CrossValidator = cv_efc8cf15a3ab {als_10a2c5c69e40-rank: 12,als_10a2c5c69e40-regParam: 0.05 } {als_10a2c5c69e40-rank: 20,als_10a2c5c69e40-regParam: 0.05 } {als_10a2c5c69e40-rank: 12,als_10a2c5c69e40-regParam: 0.15 } {als_10a2c5c69e40-rank: 20,als_10a2c5c69e40-regParam: 0.15 }由以上輸出內容可以看出，共打印了四組參數組合。
18、使用找到的最優模型來再次進行預測，并對預測結果進行評估。在notebook單元格中，輸入以下代碼：
1. //這需要一段時間才能完成超過10個實驗 2. val cvModel = crossValidator.fit(trainingData) 3. 4. // 執行預測并刪除空值 5. val predictions2 = cvModel.transform(testData).na.drop 6. 7. val evaluator2 = new RegressionEvaluator() 8. .setMetricName("rmse") 9. .setLabelCol("rating") 10. .setPredictionCol("prediction") 11. 12. val rmse2 = evaluator2.evaluate(predictions2)同時按下【Shift+Enter】鍵，執行以上代碼，輸出內容如下：
cvModel: org.apache.spark.ml.tuning.CrossValidatorModel = cv_efc8cf15a3ab predictions2: org.apache.spark.sql.DataFrame = [userId: int, movieId: int … 2 more fields] evaluator2: org.apache.spark.ml.evaluation.RegressionEvaluator = regEval_4dd08e13c0e9 rmse2: Double = 0.9471342462991672由以上輸出內容可以看出，rmse2的值要低于rmse1，預測結果相比之前更加準確。
總結
以上是生活随笔為你收集整理的基于Spark的电影推荐系统实现的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 湖北省汉十高速公路项目接近尾声，所想所感
- 下一篇： IDEA web项目导出 war 包