Spark入門(五)Spark的reduce和reduceByKey

NO IMAGE

reduce和reduceByKey的區別

reduce和reduceByKey是spark中使用地非常頻繁的,在字數統計中,可以看到reduceByKey的經典使用。那麼reduce和reduceBykey的區別在哪呢?reduce處理數據時有著一對一的特性,而reduceByKey則有著多對一的特性。比如reduce中會把數據集合中每一個元素都處理一次,並且每一個元素都對應著一個輸出。而reduceByKey則不同,它會把所有key相同的值處理並且進行歸併,其中歸併的方法可以自己定義。

例子

在單詞統計中,我們採用的就是reduceByKey,對於每一個單詞我們設置成一個鍵值對(key,value),我們把單詞作為key,即key=word,而value=1,因為遍歷過程中,每個單詞的出現一次,則標註1。那麼reduceByKey則會把key相同的進行歸併,然後根據我們定義的歸併方法即對value進行累加處理,最後得到每個單詞出現的次數。而reduce則沒有相同Key歸併的操作,而是將所有值統一歸併,一併處理。

spark的reduce

我們採用scala來求得一個數據集中所有數值的平均值。該數據集包含5000個數值,數據集以及下列的代碼均可從github下載,數據集名稱為”avg”。為求得這個數據集中的平均值,我們先用map對文本數據進行處理,將其轉換成long類型。

數據集內容:

Spark入門(五)Spark的reduce和reduceByKey

reduce求平均值scala實現

import org.apache.spark.{SparkConf, SparkContext}
object SparkReduce {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local").setAppName("SparkReduce")
val sc = new SparkContext(conf)
//將String轉成Long類型
val numData = sc.textFile("./avg").map(num => num.toLong)
//reduce處理每個值
println(numData.reduce((x,y)=>{
println("x:"+x)
println("y:"+y)
x+y
})/numData.count())
}
}

reduce求平均值Java實現

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
public class SparkReduceJava {
public static void main(String[] main){
SparkConf conf = new SparkConf().setAppName("SparkReduceJava").setMaster("local");
JavaSparkContext sc = new JavaSparkContext(conf);
reduceJava(sc);
reduceJava8(sc);
}
public static void reduceJava(JavaSparkContext sc){
JavaRDD<Long>textData = sc.textFile("./avg").map(new Function<String, Long>() {
@Override
public Long call(String s) throws Exception {
return Long.parseLong(s);
};
});
System.out.println(
textData.reduce(new Function2<Long, Long, Long>() {
@Override
public Long call(Long aLong, Long aLong2) throws Exception {
System.out.println("x:"+aLong);
System.out.println("y:"+aLong2);
return aLong+aLong2;
}
})/textData.count()
);
}
public static void reduceJava8(JavaSparkContext sc){
JavaRDD<Long>textData = sc.textFile("./avg").map(s->Long.parseLong(s));
System.out.println(textData.reduce((x,y)->x+y)/textData.count());
}
}

reduce求平均值python實現

from pyspark import SparkConf,SparkContext
conf = SparkConf().setMaster("local").setAppName("SparkReduce")
sc = SparkContext(conf=conf)
numData = sc.textFile("./avg").map(lambda s:int(s))
print(numData.reduce(lambda x,y:x+y)/numData.count())

運行結果

觀察運行結果,我們不難發現,x存放的是累加後的值,y是當前值,x初始為0。事實上,x正是存放上次處理的結果,而y則是本次的數值。不斷做x+y就並且放回累加後的結果作為下一次x的值。這樣就可以得
到數值總和。最後將總和除以總數就能夠得到平均值。

scala或java運行結果

平均值只保留了整數

x:222783
y:48364
x:271147
y:204950
x:476097
y:261777
x:737874
y:166827
x:904701
y:154005
x:1058706
y:150029
x:1208735
y:140158
x:1348893
y:404846
x:1753739
y:542750
...
...
平均值是:334521

Spark入門(五)Spark的reduce和reduceByKey

python運行結果

python默認保留了小數

334521.2714

Spark入門(五)Spark的reduce和reduceByKey

spark的reduceByKey

spark的reduceByKey對要處理的值進行了差別對待,只有key相同的才能進行reduceByKey,則也就要求了進行reduceByKey時,輸入的數據必須滿足有鍵有值。由於上述的avg我們是用隨機數生成的,那麼我們可以用reduceByKey完成一個其他功能,即統計隨機數中末尾是0-9各個數值出現的個數。

scala實現

import org.apache.spark.{SparkConf, SparkContext}
object SparkReduceByKey {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local").setAppName("SparkReduce")
val sc = new SparkContext(conf)
//將String轉成Long類型
val numData = sc.textFile("./avg").map(num => (num.toLong%10,1))
numData.reduceByKey((x,y)=>x+y).foreach(println(_))
}
}

java實現

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;
public class SparkReduceByKeyJava {
public static void main(String[] main){
SparkConf conf = new SparkConf().setAppName("SparkReduceJava").setMaster("local");
JavaSparkContext sc = new JavaSparkContext(conf);
reduceByKeyJava(sc);
reduceByKeyJava8(sc);
}
public static void reduceByKeyJava(JavaSparkContext sc){
JavaPairRDD<Integer,Integer> numData = sc.textFile("./avg").mapToPair(new PairFunction<String, Integer, Integer>() {
@Override
public Tuple2<Integer, Integer> call(String s) throws Exception {
return new Tuple2<Integer, Integer>(Integer.parseInt(s)%10,1);
}
});
System.out.println(numData.reduceByKey(new Function2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer integer, Integer integer2) throws Exception {
return integer+integer2;
}
}).collectAsMap());
}
public static void reduceByKeyJava8(JavaSparkContext sc){
JavaPairRDD<Integer,Integer> numData = sc.textFile("./avg").mapToPair(s->new Tuple2<>(Integer.parseInt(s)%10,1));
System.out.println(numData.reduceByKey((x,y)->x+y).collectAsMap());
}
}

python實現

from pyspark import SparkConf,SparkContext
conf = SparkConf().setMaster("local").setAppName("SparkReduce")
sc = SparkContext(conf=conf)
print(sc.textFile("./avg").map(lambda s:(int(s)%10,1)).reduceByKey(lambda x,y:x+y).collectAsMap())

運行結果

scala運行結果

(4,522)
(0,462)
(1,495)
(6,519)
(3,463)
(7,544)
(9,518)
(8,533)
(5,483)
(2,461)

Spark入門(五)Spark的reduce和reduceByKey

java運行結果

{8=533, 2=461, 5=483, 4=522, 7=544, 1=495, 9=518, 3=463, 6=519, 0=462}

Spark入門(五)Spark的reduce和reduceByKey

python運行結果

{3: 463, 4: 522, 0: 462, 7: 544, 5: 483, 9: 518, 8: 533, 6: 519, 2: 461, 1: 495}

Spark入門(五)Spark的reduce和reduceByKey

我們注意到三個程序輸出的順序不一樣,但是本質的結果都是一致的。這裡體現了spark的一個優點,由於是在單機本地上,該優點表現出來的是相同輸入輸出結果順序不同。但是在集群中,該優點表現出來的是在集群中各自處理,而後返回結果。當數量足夠大的時候,這個優點就更加明顯。

對結果進行排序

那麼為了能夠使得輸出結果順序一致,我們可以對數據進行排序後輸出,那麼這裡就涉及到了sortByKey。

scala實現

import org.apache.spark.{SparkConf, SparkContext}
object SparkReduceByKey {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local").setAppName("SparkReduce")
val sc = new SparkContext(conf)
//將String轉成Long類型
val numData = sc.textFile("./avg").map(num => (num.toLong%10,1))
//根據key排序後輸出
numData.reduceByKey((x,y)=>x+y).sortByKey().foreach(println(_))
}
}

java實現

特別注意這裡用的是collect,而不是collectMap,因為java中轉換成Map會打亂順序

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;
public class SparkReduceByKeyJava {
public static void main(String[] main){
SparkConf conf = new SparkConf().setAppName("SparkReduceJava").setMaster("local");
JavaSparkContext sc = new JavaSparkContext(conf);
reduceByKeyJava(sc);
reduceByKeyJava8(sc);
}
public static void reduceByKeyJava(JavaSparkContext sc){
JavaPairRDD<Integer,Integer> numData = sc.textFile("./avg").mapToPair(new PairFunction<String, Integer, Integer>() {
@Override
public Tuple2<Integer, Integer> call(String s) throws Exception {
return new Tuple2<Integer, Integer>(Integer.parseInt(s)%10,1);
}
});
System.out.println(numData.reduceByKey(new Function2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer integer, Integer integer2) throws Exception {
return integer+integer2;
}
}).sortByKey().collect());
}
public static void reduceByKeyJava8(JavaSparkContext sc){
JavaPairRDD<Integer,Integer> numData = sc.textFile("./avg").mapToPair(s->new Tuple2<>(Integer.parseInt(s)%10,1));
System.out.println(numData.reduceByKey((x,y)->x+y).sortByKey().collect());
}
}

python實現

from pyspark import SparkConf,SparkContext
conf = SparkConf().setMaster("local").setAppName("SparkReduce")
sc = SparkContext(conf=conf)
print(sc.textFile("./avg").map(lambda s:(int(s)%10,1)).reduceByKey(lambda x,y:x+y).sortByKey().collectAsMap())

得到結果,這裡只給出scala輸出的結果,其他輸出的結果一致,只是表現形式不同

(0,462)
(1,495)
(2,461)
(3,463)
(4,522)
(5,483)
(6,519)
(7,544)
(8,533)
(9,518)

數據集以及代碼都可以在github上下載。

相關文章

Spark實戰搭建我們的Spark分佈式架構

Spark實戰尋找5億次訪問中,訪問次數最多的人

Spark入門(七)Spark的intersection、subtract、union和distinct

Spark入門(六)Spark的combineByKey、sortBykey