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(_ != '\t')
  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('\t')
  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()