[解決済み] reduceByKeyとgroupByKeyとaggregateByKeyとcombineByKeyの違いのスパーク

質問

の違いを説明できる人はいますか？ reducebykey , groupbykey , aggregatebykey と combinebykey ? これに関するドキュメントを読みましたが、正確な違いを理解することができませんでした。

例を挙げての説明があると助かります。

どのように解決するのですか？

groupByKeyです。

構文です。

sparkContext.textFile("hdfs://")
                    .flatMap(line => line.split(" ") )
                    .map(word => (word,1))
                    .groupByKey()
                    .map((x,y) => (x,sum(y)))
            

groupByKey は、データがネットワーク経由で送信され、縮小されたワーカーに収集されるため、ディスク外の問題を引き起こす可能性があります。

reduceByKeyです。

構文です。

sparkContext.textFile("hdfs://")
                    .flatMap(line => line.split(" "))
                    .map(word => (word,1))
                    .reduceByKey((x,y)=> (x+y))

データは各パーティションで結合され、ネットワーク上に送信するのは各パーティションの1つのキーに対して1つの出力のみとなります。 reduceByKey は、すべての値をまったく同じ型を持つ別の値に結合する必要があります。

aggregateByKeyです。

と同じ reduceByKey と同じで、初期値をとります。

入力として3つのパラメータ

1. 初期値
2. コンバイナロジック
3. シーケンスオペレーションロジック

例

val keysWithValuesList = Array("foo=A", "foo=A", "foo=A", "foo=A", "foo=B", "bar=C", "bar=D", "bar=D")
    val data = sc.parallelize(keysWithValuesList)
    //Create key value pairs
    val kv = data.map(_.split("=")).map(v => (v(0), v(1))).cache()
    val initialCount = 0;
    val addToCounts = (n: Int, v: String) => n + 1
    val sumPartitionCounts = (p1: Int, p2: Int) => p1 + p2
    val countByKey = kv.aggregateByKey(initialCount)(addToCounts, sumPartitionCounts)

を出力します。 キーで集計する 結果 バー -> 3 foo -> 5

combineByKeyです。

3つのパラメータを入力として

1. 初期値：似て非なるもの aggregateByKey とは異なり、常に定数を渡す必要はなく、新しい値を返す関数を渡すことができます。
2. マージ関数
3. コンバイン機能

例

val result = rdd.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 (k,v) => (k,v._1/v._2.toDouble) })
        result.collect.foreach(println)

reduceByKey , aggregateByKey , combineByKey より優先される groupByKey

参照 groupByKey を回避する