scala-spark:如何在groupby之后过滤RDD

Question

我已经开始使用RDD，它有竖线分隔的字符串。我已经对数据进行了处理，并得到了以下格式：

((0001F46468,239394055),(7665710590658745,-414963169),0,1420276980302)
((0001F46468,239394055),(8016905020647641,183812619),1,1420347885727)
((0001F46468,239394055),(6633110906332136,294201185),1,1420398323110)
((0001F46468,239394055),(6633110906332136,294201185),0,1420451687525)
((0001F46468,239394055),(7722056727387069,1396896294),1,1420537469065)
((0001F46468,239394055),(7722056727387069,1396896294),1,1420623297340)
((0001F46468,239394055),(8045651092287275,-4814845),1,1420720722185)
((0001F46468,239394055),(5170029699836178,-1332814297),0,1420750531018)
((0001F46468,239394055),(7722056727387069,1396896294),0,1420807545137)
((0001F46468,239394055),(4784119468604853,1287554938),1,1421050087824) 

只是为了给出一个关于数据描述的高级视图。您可以认为主元组中的第一个元素(第一个元组)是用户标识，第二个元组是产品标识，第三个元素是用户对产品的偏好。(为了将来参考，我将上面的数据集标记为val userData)

我的目标是，如果用户对产品同时设置了正(1)和负(0)偏好，则只使用正面的记录。例如：

((0001F46468,239394055),(6633110906332136,294201185),1,1420398323110)
((0001F46468,239394055),(6633110906332136,294201185),0,1420451687525)

我只想留着

((0001F46468,239394055),(6633110906332136,294201185),1,1420398323110) 

所以我按照用户产品元组(0001F46468,239394055),(6633110906332136,294201185对用户进行了分组

val groupedFiltered = userData.groupBy(x => (x._1, x._2)).map(u => {
      for(k <- u._2) {
        if(k._3 > 0)
          u
      }
    })

但这会返回空的元组。

因此，我采取了以下方法：

val groupedFiltered = userData. groupBy(x => (x._1, x._2)).flatMap(u => u._2).filter(m => m._3 > 0)

((47734739656882457,-1782798434),(7585453414177905,-461779195),1,1422013413082)
((47734739656882457,-1782798434),(7585453414177905,-461779195),1,1422533237758)
((55218449094787901,-1374432022),(6227831620534109,1195766703),1,1420410603596)
((71212122719822610,-807015489),(6769904840922490,1642054117),1,1422549467554)
((75414197560031509,1830213715),(6724015489416254,-1389654186),1,1420196951100)
((60422797294995441,734266951),(6335216393920738,1528026712),1,1421161253600)
((35091051395844216,451349158),(8135854751464083,-1751839326),1,1422083101033)
((16647193023519619,990937787),(5384884550662007,-910998857),1,1420659873572)
((43355867025936022,-945669937),(7336240855866885,518993644),1,1420880078266)
((12188366927481231,-2007889717),(5336507724485344,363519858),1,1420827788022)

这是有希望的，但看起来它采用了所有的记录，其中我只想要，如果用户有1和0，对于相同的项目，只保留1的记录。

Answer 1

您只能保留分组结果中的最大用户首选项。

userData
 // group by user and product
 .groupBy(x => (x._1, x._2))
 // only keep the maximum user preference per user/product
 .mapValues(_.maxBy(_._3))
 // only keep the values
 .values