object MusicRecommend {
    def musicRecommend()={
        val conf = new SparkConf().setAppName("musicRecommend")
        val sc = new SparkContext(conf)
//      處理藝術家資料 
        val rawArtistData = sc.textFile("ds/artist", 4)
//      由於檔案存在非法行 不是用\t分割的
        /*val aritstByID = rawArtistData.map{ l => 
            val (id,name) = l.span(_ != "\t")
            (id.toInt,name.trim())  
        }*/
 


        val artistByID = rawArtistData.flatMap{line =>
            val (id,name) = line.span(_ != '\t')    
            if(name.isEmpty()) None
            else 
                try{
                    Some((id.toInt,name.trim()))
                }catch{
                    case e :Exception => None
                }
        }
//      處理藝術家別名資料
 

        val rawArtistAlias = sc.textFile("ds/alias")
        val artistAlias = rawArtistAlias.flatMap{ line =>
            val tokens=line.trim().split("\\s+")
            if(tokens.length != 2){
                None
            }else{
                Some((tokens(0).toInt,tokens(1).toInt))
            }
        }.collectAsMap()
//      這裡使用廣播變數,spark執行一個stage階段時,會為執行函式建立一個閉包，
 

//      也就是所有任務需要的資訊的2進位制形式。閉包包含了函式引用的所有資料結構。
//      spark把這個閉包傳送到叢集上每個executor上.
//      當許多工使用同一個不可變的資料結構時，我們應該使用廣播變數。
//      好處是可以在多個作業和階段stage之間快取資料
        val bArtistAlias = sc.broadcast(artistAlias)
//      處理使用者收聽記錄
        val rawUserArtistData = sc.textFile("ds/user*")
//      生成訓練資料
        val trainData = rawUserArtistData.map{ line =>
            val Array(userID,artistID,count) = line.split("\\s+").map(_.toInt)
            val finalArtistID=bArtistAlias.value.getOrElse(artistID, artistID)
            Rating(userID,finalArtistID,count)
        }.cache()
//      使用訓練資料進行訓練
//      MatrixFactorizationModel 其他的引數都是超引數 其值直接影響到最終結果
        val model = ALS.trainImplicit(trainData, 10, 5, 0.01, 1.0)
//      檢查推薦結果是否合理
        val userId = 2093760
        val rcNum = 5
//      1.獲取使用者2093760的資料
        val rawArtistsForUser = rawUserArtistData.map(_.split("\\s+"))
            .filter{case Array(user,_,_) => user.toInt == userId}
//      2.獲取使用者2093760感興趣的藝術家集
        val existingProducts = rawArtistsForUser.map {case Array(_,artist,_) => artist.toInt}
            .collect.toSet
//      3.找到該藝術家對應的名稱，並列印
        artistByID.filter{case (id,name) =>
                    existingProducts.contains(id)
            }.values.collect.foreach(println)
//      獲取該使用者的幾個推薦結果
        val recommendations = model.recommendProducts(userId, rcNum)
//      列印推薦結果
        recommendations.foreach(println)
//      結果構成Rating(userid,artistID,rating)rating表示一個數值，其值越接近1表示推薦結果越好

    }
}