Spark運算元:transformation之鍵值轉換groupByKey、reduceByKey、reduceByKeyLocally
阿新 • • 發佈:2018-12-11
1、groupByKey
1)def groupByKey(): RDD[(K, Iterable[V])] 2)def groupByKey(numPartitions: Int): RDD[(K, Iterable[V])] 3)def groupByKey(partitioner: Partitioner): RDD[(K, Iterable[V])]
該函式用於將RDD[K,V]中每個K對應的V值,合併到一個集合Iterable[V]中,引數numPartitions指分割槽數,partitioner指分割槽函式。
scala> var rdd1 = sc.makeRDD(Array(("A",0),("A",2),("B",1),("B",2),("C",1))) rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[89] at makeRDD at :21 scala> rdd1.groupByKey().collect res81: Array[(String, Iterable[Int])] = Array((A,CompactBuffer(0, 2)), (B,CompactBuffer(2, 1)), (C,CompactBuffer(1)))
2、reduceByKey
1)def reduceByKey(func: (V, V) => V): RDD[(K, V)] 2)def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)] 3)def reduceByKey(partitioner: Partitioner, func: (V, V) => V): RDD[(K, V)]
該函式將RDD[K,V]中的每個K對應的V值根據對映函式來計算。引數numPartitions指分割槽數,partitioner指分割槽函式。
scala> var rdd1 = sc.makeRDD(Array(("A",0),("A",2),("B",1),("B",2),("C",1))) rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[91] at makeRDD at :21 scala> rdd1.partitions.size res82: Int = 15 scala> var rdd2 = rdd1.reduceByKey((x,y) => x + y) rdd2: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[94] at reduceByKey at :23 scala> rdd2.collect res85: Array[(String, Int)] = Array((A,2), (B,3), (C,1)) scala> rdd2.partitions.size res86: Int = 15 scala> var rdd2 = rdd1.reduceByKey(new org.apache.spark.HashPartitioner(2),(x,y) => x + y) rdd2: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[95] at reduceByKey at :23 scala> rdd2.collect res87: Array[(String, Int)] = Array((B,3), (A,2), (C,1)) scala> rdd2.partitions.size res88: Int = 2
3、reduceByKeyLocally:def reduceByKeyLocally(func: (V, V) => V): Map[K, V]
該函式將RDD[K,V]中每個K對應的V值根據對映函式來運算,運算結果對映到一個Map[K,V]中,而不是RDD[K,V]。
scala> var rdd1 = sc.makeRDD(Array(("A",0),("A",2),("B",1),("B",2),("C",1))) rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[91] at makeRDD at :21 scala> rdd1.reduceByKeyLocally((x,y) => x + y) res90: scala.collection.Map[String,Int] = Map(B -> 3, A -> 2, C -> 1)