ペアRDD変換

Question

私は、データはこれに似て設定している場合：

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

と出力があるべきように、Iごとのキーの平均を見つけたい：

(1, 2), (2, 3/2), (3, 2)私はこれを行うことができますgroupByKey, countByKey, and reduceByKeyをどういうふうに使うか、以下のようなcombineByKeyメソッドを使う必要があります：groupByKey, countByKey, and reduceByKeyを使ってみましたが、この組み合わせのメソッドはうまくいきません。これらの3つのメソッドを使って行う方法が分かっているのでしょうか？

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

、トリプレットへduplesの形質転換前に、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)} 

私はあなたがサムとサイズを使用することができますので、あなたはそれが配列の値を置くgroupByKey際に忘れてしまった