Spark大数据分析与实战：基于Spark MLlib 实现音乐推荐

基于Spark MLlib 实现音乐推荐

一、实验背景：

熟悉 Audioscrobbler 数据集

基于该数据集选择合适的 MLlib 库算法进行数据处理

进行音乐推荐（或用户推荐）

二、实验目的：

计算AUC评分最高的参数

利用AUC评分最高的参数，给用户推荐艺术家

对多个用户进行艺术家推荐

利用AUC评分最高的参数，给艺术家推荐喜欢他的用户

三、实验步骤：

安装Hadoop和Spark
启动Hadoop与Spark
将文件上传到 HDFS
实现音乐推荐

四、实验过程：

1、安装Hadoop和Spark

具体的安装过程在我以前的博客里面有，大家可以通过以下链接进入操作：

提示：如果IDEA未构建Spark项目，可以转接到以下的博客：

IDEA使用Maven构建Spark项目：https://blog.csdn.net/weixin_/article/details/

2、启动Hadoop与Spark

查看3个节点的进程

master
slave1

slave2
在这里插入图片描述

3、将文件上传到 HDFS

Shell命令：

[root@master ~]# cd /opt/data/profiledata_06-May-2005/ [root@master profiledata_06-May-2005]# ls [root@master profiledata_06-May-2005]# hadoop dfs -put artist_alias.txt artist_data.txt user_artist_data.txt /spark/input

在这里插入图片描述

4、实现音乐推荐

源代码：

package com.John.SparkProject import org.apache.spark.SparkConf import org.apache.spark.broadcast.Broadcast import org.apache.spark.ml.recommendation.{ 
    ALS, ALSModel} import org.apache.spark.sql.functions._ import org.apache.spark.sql.{ 
    DataFrame, Dataset, SparkSession} import scala.collection.mutable.ArrayBuffer import scala.util.Random / * @author John * @Date 2021/5/25 12:49 */ object project02 { 
     def main(args: Array[String]): Unit = { 
     / * 前期环境配置以及数据准备 */ // 创建一个SparkSession对象 val conf = new SparkConf() .setMaster("local[2]") .setAppName("Project02_RecommenderApp") .set("spark.sql.crossJoin.enabled", "true") val spark = new SparkSession.Builder() .config(conf) .getOrCreate() import spark.implicits._ // 导入 artist_data.txt 文件 (每个艺术家的 ID 和对应的名字) // 字段名分别是: artistid artist_name val rawArtistData = spark.read.textFile("hdfs://192.168.254.122:9000/spark/input/artist_data.txt") val artistIdDF = transformArtistData(rawArtistData) val artistIdDFtest = transformArtistData1(rawArtistData) // 导入 artist_alias.txt 文件 (将拼写错误的艺术家 ID 或ID 变体对应到该艺术家的规范 ID) // 字段名分别是: badid goodid val rawAliasData = spark.read.textFile("hdfs://192.168.254.122:9000/spark/input/artist_alias.txt") val artistAlias = transformAliasData(rawAliasData).collect().toMap // 导入 user_artist_data.txt 文件 (用户音乐收听数据) // 字段名分别是 userid artistid playcount val rawUserArtistData = spark.read.textFile("hdfs://192.168.254.122:9000/spark/input/user_artist_data.txt") // 整合数据 val allDF = transformUserArtistData(spark, rawUserArtistData, artistAlias) allDF.persist() // 拆分训练集和测试集 val Array(trainDF, testDF) = allDF.randomSplit(Array(0.9, 0.1)) trainDF.persist() // 查看一下指定 user 听过的艺术家 allDF.join(artistIdDFtest,"artist").select("name").filter("user=''").show(5) / * 根据题目要求参数 * 给用户推荐艺术家 */ // // 根据题目要求构建模型 // val als = new ALS() // .setSeed(Random.nextLong()) // .setImplicitPrefs(true) // .setRank(10) // 模型潜在因素个数 // .setRegParam(0.01) // 正则化参数 // .setAlpha(1.0) // 管理偏好观察值的 基线置信度 // .setMaxIter(5) // 最大迭代次数 // .setUserCol("user") // .setItemCol("artist") // .setRatingCol("count") // .setPredictionCol("prediction") // // 用训练数据训练模型 // val model = als.fit(trainDF) // // 释放缓存资源 // trainDF.unpersist() // // 开始推荐 // val userID =  // val artistNum = 5 // val recommendDF = recommend(model, userID, artistNum, artistIdDF) // // val strings = recommendDF.map(_.mkString("|")).collect() // println(strings.toBuffer) / * 计算AUC评分最高的参数 * 原理:循环指定参数暴力计算，根据AUC计算出评分最高的参数进行建模 */ // // 艺术家id数据，用于AUC评分 // val allArtistIds = allDF.select("artist").as[Int].distinct().collect() // val bAllArtistIds = spark.sparkContext.broadcast(allArtistIds) // // // 网格搜索 // val evaluations = // // 利用for循环，生成不同的超参数配置 // for (rank <- Seq(5, 30); // regParam <- Seq(4.0, 0.0001); // alpha <- Seq(1.0, 40.0)) // yield { 
     // // 构建模型 // val als = new ALS() // .setSeed(Random.nextLong()) // .setImplicitPrefs(true) // .setRank(rank) // .setRegParam(regParam) // .setAlpha(alpha) // .setMaxIter(5) // .setUserCol("user") // .setItemCol("artist") // .setRatingCol("count") // .setPredictionCol("prediction") // val model = als.fit(trainDF) // val auc = areaUnderCurve(testDF, bAllArtistIds, model.transform) // // 释放资源 // model.userFactors.unpersist() // model.itemFactors.unpersist() // (auc, (rank, regParam, alpha)) // } // // 按评分降序输出各参数信息 // evaluations.sorted.reverse.foreach(println) // (0.77203,(30,4.0,40.0)) // 最优参数 // (0.63009,(30,4.0,1.0)) // (0.62062,(30,1.0E-4,40.0)) // (0.82363,(5,4.0,40.0)) // (0.29872,(5,1.0E-4,40.0)) // (0.71897,(5,4.0,1.0)) // (0.45514,(5,1.0E-4,1.0)) // (0.45603,(30,1.0E-4,1.0)) / * 利用AUC评分最高的参数 * 给用户推荐艺术家 */ // // 利用AUC评分最高的参数进行模型构建 // val als = new ALS() // .setSeed(Random.nextLong()) // .setImplicitPrefs(true) // .setRank(30) // 模型潜在因素个数 // .setRegParam(4.0) // 正则化参数 // .setAlpha(40.0) // 管理偏好观察值的 基线置信度 // .setMaxIter(5) // 最大迭代次数 // .setUserCol("user") // .setItemCol("artist") // .setRatingCol("count") // .setPredictionCol("prediction") // // 训练模型 // val model = als.fit(trainDF) // // 释放缓存资源 // trainDF.unpersist() // // 开始推荐 // val userID =  // val artistNum = 5 // val recommendDF = recommend(model, userID, artistNum, artistIdDF) // recommendDF.show() // // val strings = recommendDF.map(_.mkString("|")).collect() // println(strings.toBuffer) / * 用测试数据对10个用户进行推荐 */ // // 利用AUC评分最高的参数进行模型构建 // val als = new ALS() // .setSeed(Random.nextLong()) // .setImplicitPrefs(true) // .setRank(30) // 模型潜在因素个数 // .setRegParam(4.0) // 正则化参数 // .setAlpha(40.0) // 管理偏好观察值的 基线置信度 // .setMaxIter(5) // 最大迭代次数 // .setUserCol("user") // .setItemCol("artist") // .setRatingCol("count") // .setPredictionCol("prediction") // // 训练模型 // val model = als.fit(trainDF) // //用测试数据对100个用户进行推荐 // val someUsers = testDF.select("user").as[Int].distinct.take(10) // // 推荐 // someUsers.map { user => // val recommendDF = recommend(model, user, 5, artistIdDF) // val strings = recommendDF.map(_.mkString("|")).collect() // (user, strings.toBuffer) // }.foreach(println) / * 利用AUC评分最高的参数 * 给艺术家推荐喜欢他的用户 */ // //利用AUC评分最高的参数进行模型构建 // val als = new ALS() // .setSeed(Random.nextLong()) // .setImplicitPrefs(true) // .setRank(30) // 模型潜在因素个数 // .setRegParam(4.0) // 正则化参数 // .setAlpha(40.0) // 管理偏好观察值的 基线置信度 // .setMaxIter(5) // 最大迭代次数 // .setUserCol("artist") // .setItemCol("user") // .setRatingCol("count") // .setPredictionCol("prediction") // // 训练模型 // val model = als.fit(trainDF) // // 释放缓存资源 // trainDF.unpersist() // // 开始推荐 // val artistID = 930 // val userNum = 100 // val recommendDF = recommendArtist(model, artistID, userNum, allDF) // val strings = recommendDF.select("user").map(_.mkString("|")).collect() // println(strings.toBuffer) // 关闭spark spark.stop() } / * 规范艺术家的ID，合并数据，创建一个总的数据集 * * @param spark SparkSession * @param rawUserArtistDS 用户和艺术家的关系数据集 * @param artistAlias 艺术家别名id，用于补全 * @return */ def transformUserArtistData(spark: SparkSession, rawUserArtistDS: Dataset[String], artistAlias: Map[Int, Int]): DataFrame = { 
     import spark.implicits._ // 广播变量 // 广播变量主要用于在迭代中一直需要被访问的只读变量 // 它将此变量缓存在每个 executor 里，以减少集群网络传输消耗 val bArtistAlias = spark.sparkContext.broadcast(artistAlias) rawUserArtistDS.map(line => { 
     val Array(userId, artistId, count) = line.split(' ').map(_.toInt) //以空格分隔每一行，并将数据转换为int类型 val finalArtistId = bArtistAlias.value.getOrElse(artistId, artistId) // 如果有值，那就可以得到这个值，如果没有就会得到一个默认值 (userId, finalArtistId, count) }).toDF("user", "artist", "count").cache() // 设置字段名称, cache() 以指示 Spark 在 RDD 计算好后将其暂时存储在集群的内存里 } / * 将 artist_data.txt 文件的数据转化为dataframe * * @param rawArtistData Dataset类型的artist_data数据 * @return */ def transformArtistData(rawArtistData: Dataset[String]): DataFrame = { 
     // 一个隐式implicit的转换方法，转换出正确的类型，完成编译。 import rawArtistData.sparkSession.implicits._ // 将每一行数据以tap分隔开，最后将数据类型转换为dataframe rawArtistData.flatMap(line => { 
     val (id, name) = line.span(_ != '\t') try { 
     if (name.nonEmpty) Some(id.toInt, name.trim) else None } catch { 
     case _: Exception => None } }).toDF("id", "name").cache() } / * 将 artist_data.txt 文件的数据转化为dataframe * * @param rawArtistData Dataset类型的artist_data数据 * @return */ def transformArtistData1(rawArtistData: Dataset[String]): DataFrame = { 
     // 一个隐式implicit的转换方法，转换出正确的类型，完成编译。 import rawArtistData.sparkSession.implicits._ // 将每一行数据以tap分隔开，最后将数据类型转换为dataframe rawArtistData.flatMap(line => { 
     val (id, name) = line.span(_ != '\t') try { 
     if (name.nonEmpty) Some(id.toInt, name.trim) else None } catch { 
     case _: Exception => None } }).toDF("artist", "name").cache() } / * 将 artist_alias.txt 文件的数据转化为dataframe * * @param rawAliasData Dataset类型的artist_alias数据 * @return */ def transformAliasData(rawAliasData: Dataset[String]): Dataset[(Int, Int)] = { 
     // 一个隐式implicit的转换方法，转换出正确的类型，完成编译。 import rawAliasData.sparkSession.implicits._ // 将每一行数据以tap分隔开，最后将数据类型转换为dataframe rawAliasData.flatMap(line => { 
     val Array(artist, alias) = line.split('\t') try { 
     if (artist.nonEmpty) Some(artist.toInt, alias.toInt) else None } catch { 
     case _: Exception => None } }) } / * 为指定用户推荐艺术家 * * @param model 训练好的ALS模型 * @param userId 用户id * @param howMany 推荐多少个艺术家 * @param artistIdDF 艺术家id和名称的关系映射 * @return */ def recommend(model: ALSModel, userId: Int, howMany: Int, artistIdDF: DataFrame): DataFrame = { 
     import artistIdDF.sparkSession.implicits._ // 推荐操作 // 根据 userId 推荐 artistId val toRecommend = model.itemFactors .select($"id".as("artist")) // .withColumn("user", lit(userId)) // 增加一列 userId // 根据 prediction 排序，得到输入推荐艺术家的个数 val topRecommendtions = model.transform(toRecommend) .select("artist", "prediction") .orderBy($"prediction".desc) .limit(howMany) // 将artistid转换为int，并将其放入数组 val recommendedArtistIds = topRecommendtions.select("artist").as[Int].collect() // 根据 id 得到指定的那一行数据，并将数据返回 artistIdDF.filter($"id" isin (recommendedArtistIds: _*)) } / * 为指定艺术家推荐用户 * * @param model 训练好的ALS模型 * @param artistId 艺术家id * @param howMany 推荐多少个艺术家 * @param userIdDF 用户id和名称的关系映射 * @return */ def recommendArtist(model: ALSModel, artistId: Int, howMany: Int, userIdDF: DataFrame): DataFrame = { 
     import userIdDF.sparkSession.implicits._ // 推荐操作 // 根据 artistId 推荐 userId val toRecommend = model.itemFactors .select($"id".as("user")) // .withColumn("artist", lit(artistId)) // 根据 prediction 排序，得到输入推荐用户的个数 val topRecommendtions = model.transform(toRecommend) .select("user", "prediction") .orderBy($"prediction".desc) .limit(howMany) // 将userid转换为int，并将其放入数组 val recommendedArtistIds = topRecommendtions.select("user").as[Int].collect() // 根据 id 得到指定的那一行数据，并将数据返回 userIdDF.filter($"user" isin (recommendedArtistIds: _*)).dropDuplicates(Seq("user")) } / * 计算AUC评分 * * @param positiveData 测试数据 * @param bAllArtistIDs 所有的艺术家ID * @param predictFunction model.transform * @return 评分 0-1 */ def areaUnderCurve(positiveData: DataFrame, bAllArtistIDs: Broadcast[Array[Int]], predictFunction: DataFrame => DataFrame): Double = { 
     import positiveData.sparkSession.implicits._ // What this actually computes is AUC, per user. The result is actually something // that might be called "mean AUC". // Take held-out data as the "positive". // Make predictions for each of them, including a numeric score val positivePredictions = predictFunction(positiveData.select("user", "artist")). withColumnRenamed("prediction", "positivePrediction") // BinaryClassificationMetrics.areaUnderROC is not used here since there are really lots of // small AUC problems, and it would be inefficient, when a direct computation is available. // Create a set of "negative" products for each user. These are randomly chosen // from among all of the other artists, excluding those that are "positive" for the user. val negativeData = positiveData.select("user", "artist").as[(Int, Int)]. groupByKey { 
     case (user, _) => user }. flatMapGroups { 
     case (userID, userIDAndPosArtistIDs) => val random = new Random() val posItemIDSet = userIDAndPosArtistIDs.map { 
     case (_, artist) => artist }.toSet val negative = new ArrayBuffer[Int]() val allArtistIDs = bAllArtistIDs.value var i = 0 // Make at most one pass over all artists to avoid an infinite loop. // Also stop when number of negative equals positive set size while (i < allArtistIDs.length && negative.size < posItemIDSet.size) { 
     val artistID = allArtistIDs(random.nextInt(allArtistIDs.length)) // Only add new distinct IDs if (!posItemIDSet.contains(artistID)) { 
     negative += artistID } i += 1 } // Return the set with user ID added back negative.map(artistID => (userID, artistID)) }.toDF("user", "artist") // Make predictions on the rest: val negativePredictions = predictFunction(negativeData). withColumnRenamed("prediction", "negativePrediction") // Join positive predictions to negative predictions by user, only. // This will result in a row for every possible pairing of positive and negative // predictions within each user. val joinedPredictions = positivePredictions.join(negativePredictions, "user"). select("user", "positivePrediction", "negativePrediction").cache() // Count the number of pairs per user val allCounts = joinedPredictions. groupBy("user").agg(count(lit("1")).as("total")). select("user", "total") // Count the number of correctly ordered pairs per user val correctCounts = joinedPredictions. filter($"positivePrediction" > $"negativePrediction"). groupBy("user").agg(count("user").as("correct")). select("user", "correct") // Combine these, compute their ratio, and average over all users val meanAUC = allCounts.join(correctCounts, Seq("user"), "left_outer"). select($"user", (coalesce($"correct", lit(0)) / $"total").as("auc")). agg(mean("auc")). as[Double].first() joinedPredictions.unpersist() meanAUC } } / * 规范艺术家的ID，合并数据，创建一个总的数据集 * * @param spark SparkSession * @param rawUserArtistDS 用户和艺术家的关系数据集 * @param artistAlias 艺术家别名id，用于补全 * @return */ def transformUserArtistData(spark: SparkSession, rawUserArtistDS: Dataset[String], artistAlias: Map[Int, Int]): DataFrame = { 
     import spark.implicits._ // 广播变量 // 广播变量主要用于在迭代中一直需要被访问的只读变量 // 它将此变量缓存在每个 executor 里，以减少集群网络传输消耗 val bArtistAlias = spark.sparkContext.broadcast(artistAlias) rawUserArtistDS.map(line => { 
     val Array(userId, artistId, count) = line.split(' ').map(_.toInt) //以空格分隔每一行，并将数据转换为int类型 val finalArtistId = bArtistAlias.value.getOrElse(artistId, artistId) // 如果有值，那就可以得到这个值，如果没有就会得到一个默认值 (userId, finalArtistId, count) }).toDF("user", "artist", "count").cache() // 设置字段名称, cache() 以指示 Spark 在 RDD 计算好后将其暂时存储在集群的内存里 } / * 将 artist_data.txt 文件的数据转化为dataframe * * @param rawArtistData Dataset类型的artist_data数据 * @return */ def transformArtistData(rawArtistData: Dataset[String]): DataFrame = { 
     // 一个隐式implicit的转换方法，转换出正确的类型，完成编译。 import rawArtistData.sparkSession.implicits._ // 将每一行数据以tap分隔开，最后将数据类型转换为dataframe rawArtistData.flatMap(line => { 
     val (id, name) = line.span(_ != '\t') try { 
     if (name.nonEmpty) Some(id.toInt, name.trim) else None } catch { 
     case _: Exception => None } }).toDF("id", "name").cache() } / * 将 artist_data.txt 文件的数据转化为dataframe * * @param rawArtistData Dataset类型的artist_data数据 * @return */ def transformArtistData1(rawArtistData: Dataset[String]): DataFrame = { 
     // 一个隐式implicit的转换方法，转换出正确的类型，完成编译。 import rawArtistData.sparkSession.implicits._ // 将每一行数据以tap分隔开，最后将数据类型转换为dataframe rawArtistData.flatMap(line => { 
     val (id, name) = line.span(_ != '\t') try { 
     if (name.nonEmpty) Some(id.toInt, name.trim) else None } catch { 
     case _: Exception => None } }).toDF("artist", "name").cache() } / * 将 artist_alias.txt 文件的数据转化为dataframe * * @param rawAliasData Dataset类型的artist_alias数据 * @return */ def transformAliasData(rawAliasData: Dataset[String]): Dataset[(Int, Int)] = { 
     // 一个隐式implicit的转换方法，转换出正确的类型，完成编译。 import rawAliasData.sparkSession.implicits._ // 将每一行数据以tap分隔开，最后将数据类型转换为dataframe rawAliasData.flatMap(line => { 
     val Array(artist, alias) = line.split('\t') try { 
     if (artist.nonEmpty) Some(artist.toInt, alias.toInt) else None } catch { 
     case _: Exception => None } }) } / * 为指定用户推荐艺术家 * * @param model 训练好的ALS模型 * @param userId 用户id * @param howMany 推荐多少个艺术家 * @param artistIdDF 艺术家id和名称的关系映射 * @return */ def recommend(model: ALSModel, userId: Int, howMany: Int, artistIdDF: DataFrame): DataFrame = { 
     import artistIdDF.sparkSession.implicits._ // 推荐操作 // 查询指定用户ID的推荐艺术家所有信息 val toRecommend = model.itemFactors .select($"id".as("artist")) // .withColumn("user", lit(userId)) // 增加一列 userId // 根据 prediction 排序，得到输入推荐艺术家的个数 val topRecommendtions = model.transform(toRecommend) .select("artist", "prediction") .orderBy($"prediction".desc) .limit(howMany) // 得到需要推荐艺术家的用户id val recommendedArtistIds = topRecommendtions.select("artist").as[Int].collect() // 根据 id 得到指定的那一行数据，并将数据返回 artistIdDF.filter($"id" isin (recommendedArtistIds: _*)) } / * 为指定艺术家推荐用户 * * @param model 训练好的ALS模型 * @param artistId 艺术家id * @param howMany 推荐多少个艺术家 * @param userIdDF 用户id和名称的关系映射 * @return */ def recommendArtist(model: ALSModel, artistId: Int, howMany: Int, userIdDF: DataFrame): DataFrame = { 
     import userIdDF.sparkSession.implicits._ // 推荐操作 // 根据 userId 推荐艺术家 val toRecommend = model.itemFactors .select($"id".as("user")) // .withColumn("artist", lit(artistId)) // 根据 prediction 排序，得到输入推荐艺术家的个数 val topRecommendtions = model.transform(toRecommend) .select("user", "prediction") .orderBy($"prediction".desc) .limit(howMany) // 得到需要推荐的艺术家的id val recommendedArtistIds = topRecommendtions.select("user").as[Int].collect() // 根据 id 得到指定的那一行数据，并将数据返回 userIdDF.filter($"user" isin (recommendedArtistIds: _*)).dropDuplicates(Seq("user")) } / * 计算AUC评分 * * @param positiveData 测试数据 * @param bAllArtistIDs 所有的艺术家ID * @param predictFunction model.transform * @return 评分 0-1 */ def areaUnderCurve(positiveData: DataFrame, bAllArtistIDs: Broadcast[Array[Int]], predictFunction: DataFrame => DataFrame): Double = { 
     import positiveData.sparkSession.implicits._ // What this actually computes is AUC, per user. The result is actually something // that might be called "mean AUC". // Take held-out data as the "positive". // Make predictions for each of them, including a numeric score val positivePredictions = predictFunction(positiveData.select("user", "artist")). withColumnRenamed("prediction", "positivePrediction") // BinaryClassificationMetrics.areaUnderROC is not used here since there are really lots of // small AUC problems, and it would be inefficient, when a direct computation is available. // Create a set of "negative" products for each user. These are randomly chosen // from among all of the other artists, excluding those that are "positive" for the user. val negativeData = positiveData.select("user", "artist").as[(Int, Int)]. groupByKey { 
     case (user, _) => user }. flatMapGroups { 
     case (userID, userIDAndPosArtistIDs) => val random = new Random() val posItemIDSet = userIDAndPosArtistIDs.map { 
     case (_, artist) => artist }.toSet val negative = new ArrayBuffer[Int]() val allArtistIDs = bAllArtistIDs.value var i = 0 // Make at most one pass over all artists to avoid an infinite loop. // Also stop when number of negative equals positive set size while (i < allArtistIDs.length && negative.size < posItemIDSet.size) { 
     val artistID = allArtistIDs(random.nextInt(allArtistIDs.length)) // Only add new distinct IDs if (!posItemIDSet.contains(artistID)) { 
     negative += artistID } i += 1 } // Return the set with user ID added back negative.map(artistID => (userID, artistID)) }.toDF("user", "artist") // Make predictions on the rest: val negativePredictions = predictFunction(negativeData). withColumnRenamed("prediction", "negativePrediction") // Join positive predictions to negative predictions by user, only. // This will result in a row for every possible pairing of positive and negative // predictions within each user. val joinedPredictions = positivePredictions.join(negativePredictions, "user"). select("user", "positivePrediction", "negativePrediction").cache() // Count the number of pairs per user val allCounts = joinedPredictions. groupBy("user").agg(count(lit("1")).as("total")). select("user", "total") // Count the number of correctly ordered pairs per user val correctCounts = joinedPredictions. filter($"positivePrediction" > $"negativePrediction"). groupBy("user").agg(count("user").as("correct")). select("user", "correct") // Combine these, compute their ratio, and average over all users val meanAUC = allCounts.join(correctCounts, Seq("user"), "left_outer"). select($"user", (coalesce($"correct", lit(0)) / $"total").as("auc")). agg(mean("auc")). as[Double].first() joinedPredictions.unpersist() meanAUC } }