出現回数に基づいてRDDをフィルタリングします。

Question

MLlib Ratingオブジェクトを使用した製品評価のRDDがあります。これは、（int userId、int productId、double rating）のタプルです。私は、評価があまりにも少ない製品のレビューであるRDDから要素を削除したいと考えています。

例えばRDDは、この可能性：

Rating(35, 1, 5.0) 
Rating(18, 1, 4.0) 
Rating(29, 2, 3.0) 
Rating(12, 2, 2.0) 
Rating(65, 3, 1.0) 

と、私は以下の2件の持つ任意の生成物を除去することをフィルタ処理すると、それだけで最後の格付けをフィルタリングし、最初の4つをバック与えるだろう。 （私は2以上の最低レビュー数をフィルタリングしたいが、たとえばちょうど）。

現在、私は評価の数の順に製品IDのシーケンスを出力し、このコードを持っていますが、私はそれに基づいて、主RDDからフィルタリングする方法がわからないし、それはとにかく非効率です：

val mostRated = ratings.map(_._2.product) 
         .countByValue 
         .toSeq 
         .sortBy(- _._2) 
         .map(_._1) 

Answer 1

ことはでき族基の長さが閾値（ここでは1）よりも大きい場合のProductId により、その後はそれに基づいてフィルタRDD。

case class Rating(UserId: Int, ProductId: Int, Rating: Double) 

val ratings = sc.parallelize(Seq(Rating(35, 1, 5.0), 
    Rating(18, 1, 4.0), 
    Rating(29, 2, 3.0), 
    Rating(12, 2, 2.0), 
    Rating(65, 3, 1.0))) 

val prodMinCounts = ratings.groupBy(_.ProductId). 
          filter(_._2.toSeq.length > 1). 
          flatMap(_._2) 
prodMinCounts.collect 
// res14: Array[Rating] = Array(Rating(35,1,5.0), Rating(18,1,4.0), Rating(29,2,3.0), Rating(12,2,2.0)) 

：グループ化されたRDDから結果を抽出するためにflatMapを使用します