spark-scala 在每个组级别进行计数的有效方法

Question

我有关注 rdd

(000740C7AD5274,8884165739991289,0)
(000740C7AD5274,5247914560952402,1)
(000740C7AD5274,6366183814312296,0)
(000740C7AD5274,8416039242203850,1)
(000740C7AD5274,8767784019249585,0)
(000740C7AD5274,8875366436847528,0)
(000740C7AD5274,6878583261589229,0)
(000740C7AD5274,7480419089929113,1)
(000740C7AD5274,7480419089929113,0)
(000740C7AD5274,8848143710281107,0)
(000740C7AD5274,7617664942496492,1)
(000740C7AD5274,4905980213247549,0)
(000740C7AD5274,6806506896473929,1)

这是代表 userId、productId、BuyorNot 信息。我想从这些数据中生成一组统计数据，例如。每个用户购买的商品数量和每个产品的用户数量。

我是这样开始的：

val userProduct = userProductRDD.groupBy(x => (x._1, x._2)).flatMap(k => (k._1, if (k._2._3) != 0) 1 else 0))

但这并没有给(userId, distinct_bought_count) 一些指导将有助于前进。

Answer 1

简单地说：

val userProducts: RDD[(String, Long)] = 
  userProductRDD.filter(_._3 == 1)                 // buys only
                .map { case (u, p, _, _) => (u, p) } // drop buys and price
                .distinct                          // keep distinct (user, product)
                .map { case (u, _) => (u, 1L) }    // word count problem
                .reduceByKey(_ + _)

// similarly
val productUsers: RDD[(Long, Long)] =
  userProductRDD.filter(_._3 == 1)
                .map { case (u, p, _, _) => (p, u) }
                .distinct
                .map { case (p, _) => (p, 1L) }
                .reduceByKey(_ + _)