小白实战学习Spark01-音乐推荐和Audioscrobbler数据集_音乐推荐数据集matlab

1.1 数据集

• 该数据集属于 隐式反馈数据
• user_arist_data.txt:包括141000个用户和160万个艺术家，记录了约2420万条用户播放艺术家歌曲的信息，其中包括播放次数信息
• artist_data.txt:包括每个艺术家的ID和对应的名字。
• artist_alias.txt：目的是为了将拼写错误的艺术家ID或ID变体对应到该艺术家的规范ID。

1.2 交替最小二乘推荐算法

• 协同过滤算法：根据两个用户的相似行为判断他们有相同的偏好。
• 潜在因素模型：试图通过数据相对少的未被观察到的底层原因，来解释大量用户和产品之间可观察到的交互。
• 矩阵分解模型：在数学上 该算法把用户和产品数据当成一个大矩阵A，矩阵第i行和第j行上的元素有值（在本数据集上代表了用户i播放过艺术家j的音乐）
  • 矩阵A是一个稀疏矩阵，算法将A分解成两个小矩阵X和Y的乘积。
  • 矩阵A有很多行和列，而X和Y的行很多但列很少（列数用k表示，这k个列就是潜在因素，用来解释数据中的交互关系）
  • 矩阵分解
    
  • 矩阵分解算法也叫矩阵补全算法，因为原始矩阵A是稀疏的，而乘积$X{Y}^T$是稠密即原始A中大量元素是缺失的，而算法为这些缺失元素补全了一个值。
  • 在该案例中，两个小矩阵分别有一行对应每个用户和每个艺术家，每行的每个值代表了对应模型的一个隐含特征（只有K个）；因此行表示了用户和艺术家怎样关联到这k个隐含特征，而隐含特征可能就对应偏好或类别；于是该问题就能被简化为 用户-特征矩阵 和 特征-艺术家矩阵 的乘积，该乘积可以理解为商品与属性之间的一个映射，然后按用户属性进行加权，而该乘积结果是对整个稠密的用户-艺术家相互关系矩阵的完整估计。
  • $A=X{Y}^T$通常没有确切的解，因为X和Y通常不够大，不足以完全表示A；$X{Y}^T$应该尽可能逼近A。
  • 想要同时得到X和Y的最优解是不可能的，因为X和Y事先都是未知的。
  • 如何求解X和Y？
    • 交替最小二乘法（ALS）
    • 基本做法：
      • 将未知的Y初始化为随机行向量矩阵，接着在给定A和Y的条件下求出X的最优解（$A_{i}Y(Y^{T}Y{)}^{-1}=X_i$）。
      • 两边精确相乘是做不到的，因此实现的目标就是最小化 |$A_{i}Y(Y^{T}Y{)}^{-1}-X_i$|或者最小化两个矩阵的平方误差
      • 我们通过由X计算每个$Y_j$而又可以由Y计算X,这样反复交替计算，X和Y最终会收敛到一个合适的结果。
    • 将ALS算法应用于隐性数据矩阵分解时，ALS矩阵分解要稍微复杂些，因为它不是分解A，而是分解由0和1组成的矩阵P，具体之后了解
    • ALS通过 稀疏的输入数据 在用简单的线性代数运算求最优解以及数据本身是可并行化的，这让ALS算法在大规模数据处理上的速度非常快。

1.3 准备数据

• 将上述提到的三个数据文件 放到HDFS中。
• 了解数据结构，对数据进行解析或转换

  • 当用户和产品的ID是数值型时，ALS算法实现效率会更高，Int的最大值（Int.MaxValue = 2147483647）,通过代码来观察我们的数据（user_artist_data.txt）。

    • 默认情况下，该数据集为每个HDFS块生成一个分区，将HDFS块大小设为典型的128MB或64MB,由于该文件大小为400MB，所以文件被拆分为3个或6个分区
    • 由于ALS这类机器学习算法会消耗更多的计算资源，因此减少数据块大小以增加分区个数会更好，减少数据块大小能够使Spark处理任务同时使用的处理器核数更多，因为每个核可以独立处理一个分区数据。通过在读取文本文件以后随着调用 .repartition(n)来指定一个不同于默认值的分区数。
    • 文件每行包含一个用户ID、一个艺术家ID和播放次数，用空格隔开。将每行的前两个值解析为整数：用户ID\艺术家ID，然后将其转换为包含列user和artist的DataFrame，通过agg简单计算出两列的最大值和最小值
    • 最大的用户ID和艺术家ID分别为 2443548 和 10794401 这两个远小于Int.Max，所以可以对其进行转换为整型。

    val rawUserArtistData = spark.read.textFile("hdfs://...")
    val userArtistDF = rawUserArtistData.map{ line =>
    	val Array(user, artist, _*) = line.split(' ')
    	(user.toInt, artist.toInt)
    }.toDF("user","artist")
    userArtistDF.agg(min("user"),max("user"),min("artist"),max("artist")).show()
    1
    2
    3
    4
    5
    6

  • 读取第二份文件（artist_data.txt）,它是难以分辨的数字ID所对应的艺术家的名字，文件包含了用制表符分割的艺术家ID和艺术家的名字

    • 由于数据中有些行不小心加入了换行符或者没有制表符，因此会直接读取时会导致NumberFormatException，这些是不应该有输出结果的。然而map()函数是要求对每个输入必须要严格返回一个值，因此这里不能用map()函数，如果用filter()方法删除那些无法解析的行，则会重复解析逻辑。
    • 这里 使用flatMap()函数，它可以将每个元素映射为零个、一个或多个结果，它将每个输入对应的零个或多个结果组成的集合简单展开，然后放入到一个更大的数据集中，在这个案例读取中搭配使用Some和None…
    • 返回一个DataFrame，艺术家ID和名字分别对应列 “id”和“name”

    		val rawArtistData = spark.read.textFile("hdfs://..")
    		val artistByID = rawArtistData.flatMap{ line =>
    			 val (id,name) = line.span(_!= '\t')
    			if (name.isEmpty){
    				 None
    				 } else {
    					try{
    					    Some((id.toInt,name.trim))
     					}catch{
     						case _:NumberFormatException => None
    				    }
     				}
    		 }.toDF("id","name")
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13

  • 读取第三份文件（artist_alias.txt）,它将拼写错误的艺术家ID或非标准化的艺术家ID映射为艺术家的正规名字。其中每行有两个ID，用制表符分隔，将“不良的”艺术家ID映射到“良好的”ID，而不是简单地把它作为包含艺术家ID二元组的数据集。

    val rawArtistAlias = spark.read.textFile("hdfs://...")
    val artistAlias = rawArtistAlias.flatMap{ line =>
    	val Array(artist,alias) = line.split("\t")
    	if (artist.isEmpty){
    		None
    	}else{
    		Some((artist.toInt,alias.toInt))
    	}
    }.collect().toMap
    1
    2
    3
    4
    5
    6
    7
    8
    9

1.4 构建第一个模型

• 如果艺术家ID存在一个不同的正规ID,我们要用artist_alias.txt将所有的艺术家ID转换为正规ID.
  • 因为Spark集群中的每个executor都要使用到artistAlias这个变量，这时可以为其创建一个广播变量，取名为bArtistAlias，使用广播变量时，Spark对集群中每个 executor只发送一个副本，并且在内存中也只保存了一个副本，能够节省大量的网络流量和内存。
  • 调用cache()让Spark在DataFrame计算好之后将其暂时存储在集群的内存里，因为ALS算法是迭代的，因此缓存是非常有益和节省计算时间。一般默认数据是序列化为字节的形式存储在内存中的，而不是对象的形式。
  • 结果会有所不同，原因是 最终的模型取决于初始特征向量，而这些初始特征向量是随机选择的，MLlib的ALS模型和其他组件默认设置了固定的随机种子，每次都会做出相同的随机选择，通过setSeed(Random.nextLong())就可以设置一个真正的随机种子。

  import org.apache.spark.ml.recommendation._
  import scala.util.Random
  
  val model = 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").
  	 fit(trainData)
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15

1.5 逐个检查推荐结果

• 首先观测模型给出的艺术家推荐直观上是否合理，通过观察某个用户的播放记录了解其品味。
• 提取该用户收听过的艺术家ID并打印他们的名字，这意味着先在输入数据中搜索该用户收听过的艺术家的ID，然后用这些ID对艺术家集合进行过滤，这样就可以获取并按序打印这些艺术家的名字

  val userID = 2093760
  val existingArtistIDs = trainData
  	.filter($"user" === userID) //找到用户2093760对应的行
  	.select("artist").as[Int].collect() //收集艺术家ID的整型集合
  artistByID.filter($"id" isin (existingArtistIDs:_*)).show() //过滤艺术家；_*变长参数语法
  1
  2
  3
  4
  5

  
• 上述 代码中只是返回艺术家们的名字，而没有返回艺术家的评分，而下述代码中则返回了评分的分值

 def makeRecommendations(
 	model: ALSModel,
 	userID: Int,
 	howMany: Int): DataFrame ={
 	val toRecommend = model.itemFactors.
 		select($"id".as("artist")).
 		withColumn("user",lit(userID)) //选择所有艺术家ID与对应的目标用户ID
 		model.transform(toRecommend).
 			select("artist","prediction").
 			orderBy($"prediction".desc).
 			limit(howMany) //对所有艺术家评分，并返回其中分值最高的
 	}
1
2
3
4
5
6
7
8
9
10
11
12

• 这种方式计算需要点时间，适用于批量评分，也不适用于实时评分
• 这类ALS算法，预测时一个0~1的模糊值，值越大，推荐质量越好。

	val topRecommendations = makeRecommendations(model,userID,5)
	topRecommendations.show()
1
2

• 将所得到的推荐艺术家的ID用类似的方法查找到艺术家的名字

	val recommendArtistIDs= topRecommendations.select("artist").as[Int].collect()
	artistByID.filter($"id" isin (recommendedArtistIDs:_*)).show()
1
2

1.6 评价推荐质量

• 推荐引擎的作用在于向用户推荐他从来没听过的艺术家名字。
• 如何让推荐变得有用靠谱（以这个案例为例）：
  • 首先从数据集中拿出一部分艺术家的播放数据放在一边（测试集），将这些数据中的艺术家作为每个用户的优秀推荐，在整个模型的构建中并不使用这些数据，让推荐引擎对模型中所有产品进行评分，然后对比检查放在一边的艺术家的推荐排行情况。
  • 通过比较放在一边的艺术家推荐排名和整个数据集合中的艺术家的推荐排名，来计算推荐引擎的的得分（对比组合中放在一边的艺术家排名高的组合所占比例即模型的得分）
• ROC曲线：接受者操作特征曲线
• AUC：ROC曲线下区域的面积，在这里把AUC看出是随机选择的好推荐比随机选择的差推荐的排名高的概率
  • AUC指标可用于评价分类器，推荐引擎为每个用户计算AUC并取其平均值，最后的结果指标稍有不同，可称为平均AUC
  • AUC是一种普遍和综合的测量整体模型输出变量的手段
• 其他指标（MAP）：准确率、召回率、平均准确率。它更强调排在最前面的推荐的质量。

1.7 计算AUC

• 接受一个交叉验证集和一个预测函数，交叉验证集代表每个用户对应的“正面的”或“好的”艺术家，预测函数把每个包含“用户-艺术家”对的DataFrame转换为一个同时包含“用户-艺术家”和“预测”的DataFrame，“预测”表示“用户”与“艺术家”之间关联的强度值，值越大，则代表推荐的排名越高。
• 将输入数据 分为训练集和验证集，训练集只用于训练ALS模型，验证集用于评估模型，我们将90%的数据用于训练，剩余的10%用于交叉验证

def areaUnderCurve(
      positiveData: DataFrame,
      bAllArtistIDs: Broadcast[Array[Int]],
      predictFunction: (DataFrame => DataFrame)): Double = {

    // 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")
      
	val allData = buildCounts(rawUserArtistData, bArtistAlias) //前文已经定义
    val Array(trainData, cvData) = allData.randomSplit(Array(0.9, 0.1))
    trainData.cache()
    cvData.cache()

    val allArtistIDs = allData.select("artist").as[Int].distinct().collect() //去重并收集给驱动程序
    val bAllArtistIDs = spark.sparkContext.broadcast(allArtistIDs)

	val model = new ALS().
          setSeed(Random.nextLong()).
          setImplicitPrefs(true).
          setRank(rank).setRegParam(regParam).
          setAlpha(alpha).setMaxIter(20).
          setUserCol("user").setItemCol("artist").
          setRatingCol("count").setPredictionCol("prediction").
          fit(trainData)

   areaUnderCurve(cvData, bAllArtistIDs, model.transform)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

• K折交叉验证：把数据分为k个大小差不多的子集，用k-1个子集做训练，在剩下的一个子集上做评估，我们把这个过程重复k次，每次用一个不同的子集做评估。
• 下面是一个简单的方法（与上面做对比），向每个用户推荐播放最多的艺术家（非个性化推荐，算是热门推荐）

	def predictMostListened(train: DataFrame)(allData: DataFrame): DataFrame = {
	    val listenCounts = train.groupBy("artist").
	      agg(sum("count").as("prediction")).
	      select("artist", "prediction")
	    allData.
	      join(listenCounts, Seq("artist"), "left_outer").
	      select("user", "artist", "prediction")
	  }

	val mostListenedAUC = areaUnderCurve(cvData, bAllArtistIDs, predictMostListened(trainData))
    println(mostListenedAUC)
1
2
3
4
5
6
7
8
9
10
11

• 调用函数并应用前两个参数得到一个偏应用函数，这个函数本身又带一个参数（allData）并返回预测结果。predictMostListened（sc,trainData）的返回结果是一个函数。

1.8 选择超参数

• setRank（10）：模型的潜在因素的个数，即“用户-特征”和“产品-特征”矩阵的列数，一般来说，它是矩阵的阶。
• setMaxIter（5）：矩阵分解迭代的次数
• setRegParam（0.01）：标准的过拟合参数，通常也叫作lambda，值越大越不容易产生过拟合，但值过大会降低分解的准确率
• setAlpha（1.0）：控制矩阵分解时，被观察到的“用户-产品”交互相对没被观察到的交互的权重。
• 其中 rank、regParam、alpha是模型的超参数，是构建过程本身的参数。
• 如何寻找最优的参数？最基本的方法就是尝试不同值的组合并对每个组合评估某个指标，然后挑选指标值最好的组合。

val evaluations =
      for (rank     <- Seq(5,  30);
           regParam <- Seq(1.0, 0.0001);
           alpha    <- Seq(1.0, 40.0)) //这里表示为3层嵌套for循环，rank循环里嵌套着regParam循环，里面再嵌套着alpha循环
      yield {
        val model = new ALS().
          setSeed(Random.nextLong()).
          setImplicitPrefs(true).
          setRank(rank).setRegParam(regParam).
          setAlpha(alpha).setMaxIter(20).
          setUserCol("user").setItemCol("artist").
          setRatingCol("count").setPredictionCol("prediction").
          fit(trainData)

        val auc = areaUnderCurve(cvData, bAllArtistIDs, model.transform)

        model.userFactors.unpersist() // 立即释放模型占用的资源
        model.itemFactors.unpersist()

        (auc, (rank, regParam, alpha))
      }

    evaluations.sorted.reverse.foreach(println) // 按第一个值（AUC）的降序排列并输出
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

• 虽然这种方式比较暴力，但是Spark这种框架可以利用并行计算和内存来提高速度。

1.9 产生推荐

• 该模型可以对所有用户产生推荐，可以用于批处理，批处理每隔一个小时或更短的时间为所有的用户重算模型和推荐结果，具体时间间隔取决于数据规模和集群速度。
• 但目前Spark MLlib的ALS并不支持向所有的用户给出推荐，该实现可以每次对一个用户进行推荐，这样每一次都会启动一个短的几秒钟的分布式作业，适合对小用户群体快速重算推荐。

1.10 总结

• ALS不是唯一的推荐算法，但是Spark MLlib目前唯一支持的算法。
• 对于显示数据，MLlib支持ALS的变体（只需要ALS用setImplicitPrefs(false)）配置，它适用于给出评分数据而不是次数数据，只需要简单使用均方根误差（RMSE）度量标准就能够评价推荐算法。
  
  
  
  
  
  

• 关注「一个热爱学习的计算机小白」公众号 ，发送「音乐推荐+数据集」获取文章中音乐推荐源码（本小白全中文注释）。

•