對RDD轉換

Question

如果我有一個數據集與此類似：

val list = List ((1,1), (1,2), (1,3), (2,2), (2,1), (3,1), (3,3))

而且我想找到每個鍵的平均所以輸出應該是：

(1, 2), (2, 3/2), (3, 2)我能做到這一點使用groupByKey, countByKey, and reduceByKey莫名其妙或我必須使用類似於下面的示例combineByKey方法：我嘗試使用groupByKey, countByKey, and reduceByKey但這種方法的組合不起作用，我想知道是否有人知道使用這三種方法做到這一點？

val result = input.combineByKey(
(v) => (v, 1), 
(acc: (Int, Int), v) => (acc._1 + v, acc._2 + 1), 
(acc1: (Int, Int), acc2: (Int, Int)) => (acc1._1 + acc2._1, acc1._2 + acc2._2)). 
map{ case (key, value) => (key, value._1/value._2.toFloat) } 

result.collectAsMap().map(println(_)) 

Answer 1

你應該嘗試以下操作：

val sc: SparkContext = ... 
val input = sc.parallelize(List((1,1), (1,2), (1,3), (2,2), (2,1), (3,1), (3,3))) 
val averages = input.groupByKey.map { case (key, values) => 
    (key, values.sum/values.size.toDouble) 
} 

println(averages.collect().toList) // List((1,2.0), (2,1.5), (3,2.0)) 

Answer 2

那麼你可以簡單地使用PairRDDFunctions.groupByKey並計算你想要什麼。

val avgKey = input.groupByKey.map{ 
    case (k, v) => (k, v.sum.toDouble/v.size) 
} 
avgkey.collect 
//res2: Array[(Int, Double)] = Array((3,2.0), (1,2.0), (2,1.5)) 

Answer 3

使用reduceByKey，與二倍體之前轉化爲三胞胎，

rdd.map{ case(k,v) => (k,(v,1)) }. 
    reduceByKey((a,v) => (a._1+v._1, a._2+v._2)). 
    map {case (k,v) => (k, v._1/v._2)}