Spark 实践——音乐推荐和 Audioscrobbler 数据集

本文基于《Spark 高级数据分析》第3章 用音乐推荐和Audioscrobbler数据
完整代码见 https://github.com/libaoquan95/aasPractice/tree/master/c3/recommend

1.获取数据集

本 章 示 例 使 用 Audioscrobbler 公 开 发 布 的 一 个 数 据 集。 Audioscrobbler 是 last.fm 的 第一个音乐推荐系统。 last.fm 创建于 2002 年，是最早的互联网流媒体广播站点之一。

Audioscrobbler 数据集有些特别， 因为它只记录了播放数据，主要的数据集在文件 user_artist_data.txt 中，它包含 141 000 个用户和 160 万个艺术家，记录了约 2420 万条用户播放艺术家歌曲的信息，其中包括播放次
数信息。

数据集在 artist_data.txt 文件中给出了每个艺术家的 ID 和对应的名字。请注意，记录播放信息时，客户端应用提交的是艺术家的名字。名字如果有拼写错误，或使用了非标准的名称， 事后才能被发现。 比如，“The Smiths”“Smiths, The”和“the smiths”看似代表不同艺术家的 ID，但它们其实明显是指同一个艺术家。因此，为了将拼写错误的艺术家 ID 或ID 变体对应到该艺术家的规范 ID，数据集提供了 artist_alias.txt 文件。

下载地址：

1. http://www-etud.iro.umontreal.ca/~bergstrj/audioscrobbler_data.html （原书地址，已失效）
2. https://github.com/libaoquan95/aasPractice/tree/master/c3/profiledata_06-May-2005（数据集大于git上传限制，分卷压缩）

2.数据处理

加载数据集

val dataDirBase = "profiledata_06-May-2005/"
val rawUserArtistData = sc.read.textFile(dataDirBase + "user_artist_data.txt")
val rawArtistData = sc.read.textFile(dataDirBase + "artist_data.txt")
val rawArtistAlias = sc.read.textFile(dataDirBase + "artist_alias.txt")

rawUserArtistData.show()
rawArtistData.show()
rawArtistAlias.show()

格式化数据集，转换成 DataFrame

val artistByID = rawArtistData.flatMap { line =>
  val (id, name) = line.span(_ != '	')
  if (name.isEmpty()){
    None
  } else {
    try {
      Some((id.toInt, name.trim))
    } catch{
      case _: NumberFormatException => None
    }
  }
}.toDF("id", "name").cache()

val artistAlias = rawArtistAlias.flatMap { line =>
  var Array(artist, alias) = line.split('	')
  if (artist.isEmpty()) {
    None
  } else {
    Some((artist.toInt, alias.toInt))
  }
}.collect().toMap
val bArtistAlias = sc.sparkContext.broadcast(artistAlias)

val userArtistDF = rawUserArtistData.map { line =>
  val Array(userId, artistID, count) = line.split(' ').map(_.toInt)
  val finalArtistID = bArtistAlias.value.getOrElse(artistID, artistID)
  (userId, artistID, count)
}.toDF("user", "artist", "count").cache()

查看 artist 别名与实名

val (badID, goodID) = artistAlias.head
artistByID.filter($"id" isin (badID, goodID)).show()

3.利用 Spark MLlib 进行推荐

Spark MLlib 使用 ALS (交替最小二乘) 来实现协同过滤算法，该模型只需传入三元组 (用户ID, 物品ID, 评分) 就可以进行计算，需要注意，用户ID 和 物品ID必须是整型数据。

val Array(trainData, cvData) = userArtistDF.randomSplit(Array(0.9, 0.1))
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)

推荐模型已经搭建完成，不过 Spark MLlib 每次只能对单个用户进行推荐，无法进行单次的全局推荐。

val userId = 2093760
val topN = 10

val toRecommend = model.itemFactors.
  select($"id".as("artist")).
  withColumn("user", lit(userId))

val topRecommendations  = model.transform(toRecommend).
  select("artist", "prediction").
  orderBy($"prediction".desc).
  limit(topN)

// 查看推荐结果
val recommendedArtistIDs = topRecommendations.select("artist").as[Int].collect()
artistByID.join(sc.createDataset(recommendedArtistIDs).
  toDF("id"), "id").
  select("name").show()