Spark MLlib 实现音乐推荐代码解

太过爱你忘了你带给我的痛 2022-10-17 01:53 282阅读 0赞

### Spark MLlib 实现音乐推荐 ###

*   *  一、准备数据
     *   *  1.1 了解数据集
         *  1.2 读取数据
         *  1.3 数据预处理
         *   *  1.3.1 对artist\_data.txt文件处理
             *  1.3.2 对artist\_alias.txt文件操作
         *  1.4 展示特征
     *  二、 代码实现 输出推荐用户的艺术家ID和艺术家。
     *  三、评估模型算法
     *   *  1.3.1 AUC 评价算法
         *  1.3.2 对训练数据进行预测
         *  1.3.3 评估模型
     *  四、多用户推荐
     *  五、主函数main
     *  六、源代码

--------------------

[Spark MLlib 实现音乐推荐][Spark MLlib]  
[Spark：对数据实现TopN][Spark_TopN]

--------------------

## 一、准备数据 ##

下载数据集 Audioscrobbler  
![在这里插入图片描述][20210601095515415.png]  
[https://github.com/libaoquan95/aasPractice/tree/master/c3/profiledata\_06-May-2005][https_github.com_libaoquan95_aasPractice_tree_master_c3_profiledata_06-May-2005]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70]

### 1.1 了解数据集 ###

user_artist_data.txt (3 columns:userid,artistid,playcount) ， 用户 艺术家 播放记录。
    
    artist_data.txt (2 columns:artistid,artist_name) 艺术家ID，艺术家名字。
    同一个艺术家可能对应不同名字，即有不同的id。 
    
    artist_alias.txt 标识每个艺术家的别名。里面用一个唯一的id标识所有同一个artist id 列表。

### 1.2 读取数据 ###

val baseDir = "D:/test/"
        val rawUserArtistData = sc.textFile(baseDir + "user_artist_data.txt")
        val rawArtistData = sc.textFile(baseDir + "artist_data.txt")
        val rawArtistAlias = sc.textFile(baseDir + "artist_alias.txt")

### 1.3 数据预处理 ###

#### 1.3.1 对artist\_data.txt文件处理 ####

我们知道，数据中不可避免会存在缺失值问题，因此我们定义一个方法把缺失值用 NULL 代替。  
**artist\_data.txt文件(id,name)操作 艺术家ID，艺术家名字。**

/* 定义buildArtistById 建立艺术家身份名字方法，
        *  思路：
        * 1.对artist_data.txt文件(id,name)操作，
        * 2.判断名字是否缺失，是把空值和异常值用None代替
        * 3.不缺失值就把ID转为int、name转为String
      */
      def buildArtistById(rawArtistData:RDD[String])={
        
        rawArtistData.flatMap{
         line=>val(id,name) = line.span(_!='\t')
          if(name.isEmpty){
        None}
          else{
         try{
        
                  Some(id.toInt,name.trim.toString) // 把ID转为int、name转为String
              } catch {
         case e:Exception =>None}}
        }
      }

#### 1.3.2 对artist\_alias.txt文件操作 ####

**user\_artist\_data.txt (3 columns:userid,artistid,playcount)** ， 用户 艺术家 播放记录。

/* buildArtistAlias 修正艺术家的ID唯一确定别名，将错误的ID和非标准ID映射正确ID。
        * 思路：
        * 1.对artist_alias.txt文件操作；
        * 2.判断数据是否缺失，是把空值用None代替，且把所有数据转为Int；
      */
      def buildArtistAlias(rawArtistAlias:RDD[String]):Map[Int,Int]=
        rawArtistAlias.flatMap {
         line =>val tokens = line.split("\t")
          if (  tokens(0).isEmpty) {
        None}
          else {
          Some((tokens(0).toInt, tokens(1).toInt))}
        }.collectAsMap()

### 1.4 展示特征 ###

接下来实现标识每个艺术家的别名。里面用一个唯一的id标识所有同一个artist id 列表。

1.  显示 **badId --> goodId**

/* 显示 badId --> goodId
        * 1.显示artist_data.txt数据具体特征，例如平均数，方差;
        * 2.将错误的ID和非标准ID修正为正确ID，badId--->goodId;
        * 3.调用 buildArtistById 确定艺术家身份名字;
        * 4.调用 buildArtistAlias 确定艺术家的ID唯一别名;
        * 5.相比较出哪个是 badId --> goodId。
      */
      def prepare(rawUserArtistData:RDD[String],
                  rawArtistData:RDD[String],rawArtistAlias:RDD[String])={
        
        val userIdStats = rawUserArtistData.map(_.split(' ')(0).toDouble).stats()
        // stats方法会返回每一列的最大值，最小值，均值、方差、总数等
        val itemIdStats = rawUserArtistData.map(_.split(' ')(1).toDouble).stats()
        println(userIdStats+"\n ----> "+itemIdStats)
    
        val artistById = buildArtistById(rawArtistData)
        val artistAlias = buildArtistAlias(rawArtistAlias)
        val (badId,goodId) = artistAlias.head
        println("badId=>"+artistById.lookup(badId)+"\n"+
          "goodId=>"+artistById.lookup(goodId))
      }

![在这里插入图片描述][20210601102852310.png]  
2. **执行 badId --> goodId；**

/* 执行 badId --> goodId；
        把(userId,artistId,count)用户,艺术家,播放记录 全部转为Int；
        把bad_id值替换成 good_id值，并存入缓存。
      */
      def buildRating(rawUserArtistData:RDD[String],bArtistAlias:Broadcast[Map[Int,Int]])={
        
        rawUserArtistData.map{
          line=>val Array(userId,artistId,count) = line.split(' ').map(_.toInt)
          // 把 bad_id替换成 good_id
          val finalArtistId = bArtistAlias.value.getOrElse(artistId,artistId)
          Rating(userId,finalArtistId,count)
        }.cache()
      }

1.  矩阵分解  
    矩阵分解模型：把用户和产品数据当成一个大矩阵A，矩阵的第 i 行和第 j 列的元素有值，代表用户i 播放过艺术家 j 的音乐；

def unpersist(model: MatrixFactorizationModel): Unit = {
        
        model.userFeatures.unpersist() // 用户
        model.productFeatures.unpersist() // 产品
      }

--------------------

## 二、 代码实现 输出推荐用户的艺术家ID和艺术家。 ##

在预处理后的数据下，进行操作，所以要调用上面预处理的各种方法。

/*  构建模型要求完整的数据：
      * 1.因此调用 'buildArtistAlias'方法 修正艺术家的ID唯一确定别名；
      * 2.调用 'buildRating'方法，把 bad_id值替换成 good_id值,且存入缓存。
      * 3.取出一部分数据来选择模型并评估模型准确度，数据集进行测试--参数调优。
      * 4.输出推荐用户的歌曲和艺术家。
      */
      def model(sc:SparkContext,rawUserArtistData:RDD[String],
                rawArtistData:RDD[String],rawArtistAlias: RDD[String])={
        
        val bArtistAlias = sc.broadcast(buildArtistAlias(rawArtistAlias))
        val trainData = buildRating(rawUserArtistData,bArtistAlias).cache()
    
        // （1）模型调优 -- 选择超参数
        val model = ALS.trainImplicit(trainData,10,5,0.01,1.0)
        val z = model.userFeatures.mapValues(_.mkString(",")).first()
        println("**---**"*10)
        println(z)
    
        // (2)查看首先查看用户2093760所听过的歌曲，如用户 2093760
        val userId = 2093760
        val recommendations = model.recommendProducts(userId,5)  //找出sum条,ID为2093760的数据
        println(s"------- 给${userId}用户推荐艺术家 --------")
        recommendations.foreach(println)
    
        val recommendationIds = recommendations.map(_.product).toSet
        val rawArtistForUser = rawUserArtistData.map(_.split(' ')).
          filter{
         case Array(user,_,_) => user.toInt==userId  }
    
        //（3）把歌曲的ID号转为int型
        val existingProduct = rawArtistForUser.map{
        
          case Array(_,artist,_) => artist.toInt
        }.collect().toSet
    
        val artistById = buildArtistById(rawArtistData)
        println("-------- 艺术家ID转为所对应的艺术家并打印 ----------")
        artistById.filter{
        
          case (id,name) =>recommendationIds.contains(id)}.values.collect().foreach(println)
        unpersist(model)
      }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 1]

--------------------

## 三、评估模型算法 ##

### 1.3.1 AUC 评价算法 ###

*  现在的排序评估指标主要包括MAP、NDCG和AUC三个指标。NDCG考虑多指标，MAP和AUC考虑单个指标。
 *  AUC用于衡量二分类问题及机器学习算法性能的泛化能力。
 *  使用AUC判断分类器(预测模型)优劣的标准：AUC=1是完美的分类器，采用这个预测模型时，存在至少一个阈值能得出完美预测。绝大多数预测的场合，不存在完美分类器。

//  AUC 评价算法
      def areaUnderCurve(positiveData:RDD[Rating],bAllItemIDS:Broadcast[Array[Int]],
                         predicitFunction:(RDD[(Int,Int)]) => RDD[Rating]) = {
        
        val positiveUserProduct = positiveData.map(r=>(r.user,r.product))
        val positivePredicition = predicitFunction(positiveUserProduct).groupBy(_.user)
        val negativeProducts = positiveUserProduct.groupByKey().mapPartitions{
        
          userIDAndPosItemIDs=>{
        
            val rand = new Random()
            val allItemIds = bAllItemIDS.value
            userIDAndPosItemIDs.map{
        case (userId,posItemIds)=>
              val posItemIdSets = posItemIds.toSet
              val negative = new ArrayBuffer[Int]()
              var i=0
              while(i<allItemIds.size && negative.size<posItemIdSets.size){
        
                val itemId = allItemIds(rand.nextInt(allItemIds.size))
                if(!posItemIdSets.contains(itemId)){
        
                  negative += itemId
                }
                i+=1
              }
              negative.map(itemId=>(userId,itemId))
            }
          }
        }.flatMap(t=>t)
    
        val negativePredicition = predicitFunction(negativeProducts).groupBy(_.user)
        positivePredicition.join(negativePredicition).values.map{
        
          case (positiveRating,negativerating) =>
            var correct = 0
            var total = 0
            for(positive<-positiveRating;negative<-negativerating){
        
              if(positive.rating>negative.rating){
        
                correct+=1
              }
              total+=1
            }
            correct.toDouble/total
        }.mean()
      }
    }

--------------------

### 1.3.2 对训练数据进行预测 ###

// 对训练数据进行预测
      def predictMostListened(sc:SparkContext,train:RDD[Rating])
             (allData:RDD[(Int,Int)])={
        
        val bListenCount = sc.broadcast(train.map(r=>(r.product,r.rating)).reduceByKey(_+_).collectAsMap())
        allData.map{
        case (user,product) => Rating(user,product,bListenCount.value.getOrElse(product,0.0))
        }
      }

--------------------

### 1.3.3 评估模型 ###

/*  1.构建模型要求完整的数据：'buildArtistAlias'修正艺术家的ID唯一确定别名；
        2.调用 'buildRating'把 bad_id值替换成 good_id值。
        3.把 artist_alias.txt文件拆分成训练集和测试集
        4.调用 areaUnderCurve 算法模型 进行评价预测准确率
      */
      def evaluate(sc:SparkContext,rawUserArtistdata:RDD[String],rawArtistAlias:RDD[String])={
        
        // (1) 调用方法获取完整数据
        val bArtistAlias = sc.broadcast(buildArtistAlias(rawArtistAlias))
        val allData = buildRating(rawUserArtistdata,bArtistAlias)
        // （2）拆分成训练集和测试集
        val Array(traindata,cvData) = allData.randomSplit(Array(0.8,0.2))
        traindata.cache()
        cvData.cache()
        val allItemIds = allData.map(_.product).distinct().collect()
        val bAllItemIds = sc.broadcast(allItemIds)
        // AUC算法调用 predictMostListened预测方法
        val mostListenedAUC = areaUnderCurve(cvData,bAllItemIds,predictMostListened(sc,traindata)) 
        println("MostListenedAUC 评估模型准确率 : "+mostListenedAUC)
    
        val evaluations =
          for ( rank  <- Array(10,  50);
               lambda <- Array(1.0, 0.0001);
               alpha  <- Array(1.0, 40.0))
            yield {
        
              val model = ALS.trainImplicit(traindata, rank, 10, lambda, alpha)
              val auc = areaUnderCurve(cvData, bAllItemIds, model.predict)
              unpersist(model)
              ((rank, lambda, alpha), auc)
            }
        println("*-*-*"*20,"evaluations:")
        evaluations.sortBy(_._2).reverse.foreach(println)
        traindata.unpersist()
        cvData.unpersist()
      }

--------------------

## 四、多用户推荐 ##

调用这个方法，会执行一段时间15-20分钟才出现结果，大概要花费6G空间，结束后会释放空间。

/*  1.构建模型要求完整的数据：'buildArtistAlias'修正艺术家的ID唯一确定别名；
          2.调用 'buildRating'把 bad_id值替换成 good_id值；
          3.选择超参数进行ALS；
          4.对用户进行评估并推荐。
      */
      // 多用户推荐
      def recommend(sc:SparkContext,rawUserArtistData:RDD[String],rawArtistData:RDD[String],rawArtistAlias:RDD[String])={
        
        //   (1) 用前面获取的最优参数对ID为2093760用户给出推荐
        val bArtistAlias = sc.broadcast(buildArtistAlias(rawArtistAlias))
        val allData = buildRating(rawUserArtistData,bArtistAlias)
        val model = ALS.trainImplicit(allData,50,10,0.1,40)
        allData.unpersist()
        val userId = 2093760
        val recommendations = model.recommendProducts(userId,5)
        val recommendProductIds = recommendations.map(_.product).toSet
        val artistById = buildArtistById(rawArtistData)
        artistById.filter{
        case (user,name)=>recommendProductIds.contains(user)}.values.collect()
        
        //  （2）对20个用户进行推荐五首歌曲的结果。
        val somUsers = allData.map(_.user).distinct().take(20)
        val someRecommendations = somUsers.map(userId => model.recommendProducts(userId,5))
        val recom = someRecommendations.map(res => res.head.user +" -> "+res.map(_.product).mkString(","))
        println("--------对前20个用户进行推荐：")
        recom.foreach(println)
      }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 2]

--------------------

## 五、主函数main ##

def main(args: Array[String]) {
        
        val conf = new SparkConf().setAppName("Recommend").setMaster("local[2]")
        val sc = new SparkContext(conf)
        sc.setLogLevel("WARN")
        val baseDir = "D:/test/"
    //    val baseDir ="hdfs://master:8020/music/"
        val rawUserArtistData = sc.textFile(baseDir + "user_artist_data.txt")
        val rawArtistData = sc.textFile(baseDir + "artist_data.txt")
        val rawArtistAlias = sc.textFile(baseDir + "artist_alias.txt")
    //    println("------------------展示特征--------------------")
    //    prepare(rawUserArtistData, rawArtistData, rawArtistAlias)
        println("------------构建模型及评价推荐质量---------------")
        model(sc,rawUserArtistData,rawArtistData,rawArtistAlias)
    //    println("------------模型调优 -- 选择超参数---------------")
    //    evaluate(sc,rawUserArtistData,rawArtistAlias)
    //    println("------------多用户推荐---------------")
    //    recommend(sc,rawUserArtistData, rawArtistData, rawArtistAlias)
      }

## 六、源代码 ##

import scala.collection.Map
    import scala.collection.mutable.ArrayBuffer
    import scala.util.Random
    import org.apache.spark.{
        SparkConf, SparkContext}
    import org.apache.spark.broadcast.Broadcast
    import org.apache.spark.mllib.recommendation._
    import org.apache.spark.rdd.RDD
    
    object Recommend {
        
      // 主函数 main调用
      def main(args: Array[String]) {
        
        val conf = new SparkConf().setAppName("Recommend").setMaster("local[2]")
        val sc = new SparkContext(conf)
        sc.setLogLevel("WARN")
        val baseDir = "D:/test/"
    //    val baseDir ="hdfs://master:8020/music/"
        val rawUserArtistData = sc.textFile(baseDir + "user_artist_data.txt")
        val rawArtistData = sc.textFile(baseDir + "artist_data.txt")
        val rawArtistAlias = sc.textFile(baseDir + "artist_alias.txt")
    //    println("------------------展示特征--------------------")
    //    prepare(rawUserArtistData, rawArtistData, rawArtistAlias)
        println("------------构建模型及评价推荐质量---------------")
        model(sc,rawUserArtistData,rawArtistData,rawArtistAlias)
    //    println("------------模型调优 -- 选择超参数---------------")
    //    evaluate(sc,rawUserArtistData,rawArtistAlias)
    //    println("------------多用户推荐---------------")
    //    recommend(sc,rawUserArtistData, rawArtistData, rawArtistAlias)
      }
    
      /* 定义buildArtistById 建立艺术家身份名字方法，
        *  思路：
        * 1.对artist_data.txt文件(id,name)操作，
        * 2.判断名字是否缺失，是把空值和异常值用None代替
        * 3.否则就把ID转为int、name转为String
      */
      def buildArtistById(rawArtistData:RDD[String])={
        
        rawArtistData.flatMap{
         line=>val(id,name) = line.span(_!='\t')
          if(name.isEmpty){
        None}
          else{
         try{
        
                  Some(id.toInt,name.trim.toString) // 把ID转为int、name转为String
              } catch {
         case e:Exception =>None}}
        }
      }
    
      /* buildArtistAlias 修正艺术家的ID唯一确定别名，将错误的ID和非标准ID映射正确ID。
        * 思路：
        * 1.对artist_alias.txt文件操作；
        * 2.判断数据是否缺失，是把空值用None代替，且把所有数据转为Int；
      */
      def buildArtistAlias(rawArtistAlias:RDD[String]):Map[Int,Int]=
        rawArtistAlias.flatMap {
         line =>val tokens = line.split("\t")
          if (  tokens(0).isEmpty) {
        None}
          else {
          Some((tokens(0).toInt, tokens(1).toInt))}
        }.collectAsMap()
    
      /* 显示 badId --> goodId
        * 1.显示artist_data.txt数据具体特征，例如平均数，方差;
        * 2.将错误的ID和非标准ID修正为正确ID，badId--->goodId;
        * 3.调用 buildArtistById 确定艺术家身份名字;
        * 4.调用 buildArtistAlias 确定艺术家的ID唯一别名;
        * 5.相比较出哪个是 badId --> goodId。
      */
      def prepare(rawUserArtistData:RDD[String],
                  rawArtistData:RDD[String],rawArtistAlias:RDD[String])={
        
        val userIdStats = rawUserArtistData.map(_.split(' ')(0).toDouble).stats()
        // stats方法会返回每一列的最大值，最小值，均值、方差、总数等
        val itemIdStats = rawUserArtistData.map(_.split(' ')(1).toDouble).stats()
        println(userIdStats+"\n ----> "+itemIdStats)
    
        val artistById = buildArtistById(rawArtistData)
        val artistAlias = buildArtistAlias(rawArtistAlias)
        val (badId,goodId) = artistAlias.head
        println("badId=>"+artistById.lookup(badId)+"\n"+
          "goodId=>"+artistById.lookup(goodId))
      }
    
      /* 执行 badId --> goodId；
        把(userId,artistId,count)用户,艺术家,播放记录 全部转为Int；
        把bad_id值替换成 good_id值
      */
      def buildRating(rawUserArtistData:RDD[String],bArtistAlias:Broadcast[Map[Int,Int]])={
        
        rawUserArtistData.map{
          line=>val Array(userId,artistId,count) = line.split(' ').map(_.toInt)
          // 把 bad_id替换成 good_id
          val finalArtistId = bArtistAlias.value.getOrElse(artistId,artistId)
          Rating(userId,finalArtistId,count)
        }.cache()
      }
    
      def unpersist(model: MatrixFactorizationModel): Unit = {
        
        model.userFeatures.unpersist()
        model.productFeatures.unpersist()
      }
    
      /*  构建模型要求完整的数据：
      * 1.因此调用 'buildArtistAlias'方法 修正艺术家的ID唯一确定别名；
      * 2.调用 'buildRating'方法，把 bad_id值替换成 good_id值,且存入缓存。
      * 3.取出一部分数据来选择模型并评估模型准确度，数据集进行测试--参数调优。
      * 4.输出推荐用户的歌曲和艺术家。
      */
      def model(sc:SparkContext,rawUserArtistData:RDD[String],
                rawArtistData:RDD[String],rawArtistAlias: RDD[String])={
        
        val bArtistAlias = sc.broadcast(buildArtistAlias(rawArtistAlias))
        val trainData = buildRating(rawUserArtistData,bArtistAlias).cache()
    
        // （1）模型调优 -- 选择超参数
        val model = ALS.trainImplicit(trainData,10,5,0.01,1.0)
        val z = model.userFeatures.mapValues(_.mkString(",")).first()
        println("**---**"*10)
        println(z)
    
        // (2)查看首先查看用户2093760所听过的歌曲，如用户 2093760
        val userId = 2093760
        val recommendations = model.recommendProducts(userId,5)  //找出sum条,ID为2093760的数据
        println(s"------- 给${userId}用户推荐艺术家 --------")
        recommendations.foreach(println)
    
        val recommendationIds = recommendations.map(_.product).toSet
        val rawArtistForUser = rawUserArtistData.map(_.split(' ')).
          filter{
         case Array(user,_,_) => user.toInt==userId  }
    
        //（3）把歌曲的ID号转为int型
        val existingProduct = rawArtistForUser.map{
        
          case Array(_,artist,_) => artist.toInt
        }.collect().toSet
    
        val artistById = buildArtistById(rawArtistData)
        println("-------- 艺术家ID转为所对应的艺术家并打印 ----------")
        artistById.filter{
        
          case (id,name) =>recommendationIds.contains(id)}.values.collect().foreach(println)
        unpersist(model)
      }
    
    
      // 对训练数据进行预测
      def predictMostListened(sc:SparkContext,train:RDD[Rating])
             (allData:RDD[(Int,Int)])={
        
        val bListenCount = sc.broadcast(train.map(r=>(r.product,r.rating)).reduceByKey(_+_).collectAsMap())
        allData.map{
        case (user,product) => Rating(user,product,bListenCount.value.getOrElse(product,0.0))
        }
      }
    
      // 评估模型
      /*  1.构建模型要求完整的数据：'buildArtistAlias'修正艺术家的ID唯一确定别名；
          2.调用 'buildRating'把 bad_id值替换成 good_id值。
          3.把 artist_alias.txt文件拆分成训练集和测试集
          4.调用 areaUnderCurve 算法模型 进行评价预测准确率
      */
      def evaluate(sc:SparkContext,rawUserArtistdata:RDD[String],rawArtistAlias:RDD[String])={
        
        // (1) 调用方法获取完整数据
        val bArtistAlias = sc.broadcast(buildArtistAlias(rawArtistAlias))
        val allData = buildRating(rawUserArtistdata,bArtistAlias)
        // （2）拆分成训练集和测试集
        val Array(traindata,cvData) = allData.randomSplit(Array(0.8,0.2))
        traindata.cache()
        cvData.cache()
        val allItemIds = allData.map(_.product).distinct().collect()
        val bAllItemIds = sc.broadcast(allItemIds)
        val mostListenedAUC = areaUnderCurve(cvData,bAllItemIds,predictMostListened(sc,traindata)) // 调用 预测方法
        println("MostListenedAUC 评估模型准确率 : "+mostListenedAUC)
    
        val evaluations =
          for ( rank  <- Array(10,  50);
               lambda <- Array(1.0, 0.0001);
               alpha  <- Array(1.0, 40.0))
            yield {
        
              val model = ALS.trainImplicit(traindata, rank, 10, lambda, alpha)
              val auc = areaUnderCurve(cvData, bAllItemIds, model.predict)
              unpersist(model)
              ((rank, lambda, alpha), auc)
            }
        println("*-*-*"*20,"evaluations:")
        evaluations.sortBy(_._2).reverse.foreach(println)
        traindata.unpersist()
        cvData.unpersist()
      }
    
      //  AUC 评价算法
      def areaUnderCurve(positiveData:RDD[Rating],bAllItemIDS:Broadcast[Array[Int]],
                         predicitFunction:(RDD[(Int,Int)]) => RDD[Rating]) = {
        
        val positiveUserProduct = positiveData.map(r=>(r.user,r.product))
        val positivePredicition = predicitFunction(positiveUserProduct).groupBy(_.user)
        val negativeProducts = positiveUserProduct.groupByKey().mapPartitions{
        
          userIDAndPosItemIDs=>{
        
            val rand = new Random()
            val allItemIds = bAllItemIDS.value
            userIDAndPosItemIDs.map{
        case (userId,posItemIds)=>
              val posItemIdSets = posItemIds.toSet
              val negative = new ArrayBuffer[Int]()
              var i=0
              while(i<allItemIds.size && negative.size<posItemIdSets.size){
        
                val itemId = allItemIds(rand.nextInt(allItemIds.size))
                if(!posItemIdSets.contains(itemId)){
        
                  negative += itemId
                }
                i+=1
              }
              negative.map(itemId=>(userId,itemId))
            }
          }
        }.flatMap(t=>t)
    
        val negativePredicition = predicitFunction(negativeProducts).groupBy(_.user)
        positivePredicition.join(negativePredicition).values.map{
        
          case (positiveRating,negativerating) =>
            var correct = 0
            var total = 0
            for(positive<-positiveRating;negative<-negativerating){
        
              if(positive.rating>negative.rating){
        
                correct+=1
              }
              total+=1
            }
            correct.toDouble/total
        }.mean()
      }
    
      /*  1.构建模型要求完整的数据：'buildArtistAlias'修正艺术家的ID唯一确定别名；
         2.调用 'buildRating'把 bad_id值替换成 good_id值；
         3.选择超参数进行ALS；
         4.对用户进行评估并推荐。
     */
      // 多用户推荐
      def recommend(sc:SparkContext,rawUserArtistData:RDD[String],rawArtistData:RDD[String],rawArtistAlias:RDD[String])={
        
        //   (1) 用前面获取的最优参数对ID为2093760用户给出推荐
        val bArtistAlias = sc.broadcast(buildArtistAlias(rawArtistAlias))
        val allData = buildRating(rawUserArtistData,bArtistAlias)
        val model = ALS.trainImplicit(allData,50,10,0.1,40)
        allData.unpersist()
        val userId = 2093760
        val recommendations = model.recommendProducts(userId,6)
        val recommendProductIds = recommendations.map(_.product).toSet
        val artistById = buildArtistById(rawArtistData)
        artistById.filter{
        case (user,name)=>recommendProductIds.contains(user)}.values.collect()
        //  （2）对20个用户进行推荐五首歌曲的结果。
        val somUsers = allData.map(_.user).distinct().take(20)
        val someRecommendations = somUsers.map(userId => model.recommendProducts(userId,5))
        val recom = someRecommendations.map(res => res.head.user +" -> "+res.map(_.product).mkString(","))
        println("--------对前20个用户进行推荐：")
        recom.foreach(println)
      }
    
    
    }

[Spark MLlib]: https://blog.csdn.net/weixin_44775255/article/details/117435696
[Spark_TopN]: https://blog.csdn.net/weixin_44775255/article/details/117412442
[20210601095515415.png]: /images/20221014/50368ca17e074daaaeeb7278268a0a4c.png
[https_github.com_libaoquan95_aasPractice_tree_master_c3_profiledata_06-May-2005]: https://github.com/libaoquan95/aasPractice/tree/master/c3/profiledata_06-May-2005
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70]: /images/20221014/3ce926ffe3c448a49841b60f837237d6.png
[20210601102852310.png]: /images/20221014/356b0e2815d14ef5846f0f3959bc3b76.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 1]: /images/20221014/ce25b490cadd4899b05ea830253540a0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 2]: /images/20221014/16cfc232b0f64f33a0c63e538de4247b.png