spark实现将相同用户（key）所有item列表聚合

电玩女神 2022-03-12 12:10 226阅读 0赞

数据：用户id，itemid，分数

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70][]

代码：

import org.apache.spark.\{SparkConf, SparkContext\}

object userwatchlist \{  
  def main(args: Array\[String\]): Unit = \{  
    val conf = new SparkConf()  
    //conf.setMaster("local")  
    conf.setAppName("userwatchlist test")

val sc = new SparkContext(conf)  
    val input\_path = sc.textFile("/train\_new.data")  
    val output\_path = "/userwatchlist\_output"

val data = input\_path.filter\{ x =>  
      val fields = x.split("\\t")  
      fields(2).toDouble > 2  
    \}.map \{ x =>  
      val fields = x.split("\\t")  
      (fields(0).toString, (fields(1).toString, fields(2).toString))  
    \}.groupByKey(10).map \{ x =>  
      val userid = x.\_1  
      val item\_score\_tuple\_list = x.\_2  
      val tmp\_arr = item\_score\_tuple\_list.toArray.sortWith(\_.\_2>\_.\_2)  
      var watchlen = tmp\_arr.length  
      if (watchlen > 5)\{  
        watchlen = 5  
      \}

val strbuf = new StringBuilder  
      for (i <- 0 until watchlen) \{  
        strbuf ++= tmp\_arr(i).\_1  
        strbuf.append(":")  
        strbuf ++= tmp\_arr(i).\_2  
        strbuf.append(" ")  
      \}  
      userid + "\\t" + strbuf

data.saveAsTextFile(output\_path)

\}  
\}

结果：

![20190308144146404.png][]

将相同用户（key）所有item列表放在了一行

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70]: /images/20220312/b5c11f185ab344519648700760fc06fb.png
[20190308144146404.png]: /images/20220312/1fd0bfcebdd849a3bf37059b542dcf2b.png