1、非lambda實現的java spark wordcount程式

public class WordCount {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("appName").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);

        //JavaPairRDD<LongWritable, Text> inputRDD = sc.hadoopFile("hdfs://master:9999/user/word.txt",
        //        TextInputFormat.class, LongWritable.class, Text.class);

        JavaRDD<String> inputRDD = sc.textFile("file:///Users/tangweiqun/test.txt");

        JavaRDD<String> wordsRDD = inputRDD.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String s) throws Exception {
                return Arrays.asList(s.split(" ")).iterator();
            }
        });

        JavaPairRDD<String, Integer> keyValueWordsRDD
                = wordsRDD.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                return new Tuple2<String, Integer>(s, 1);
            }
        });

        JavaPairRDD<String, Integer> wordCountRDD =
                keyValueWordsRDD.reduceByKey(new HashPartitioner(2),
                        new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer a, Integer b) throws Exception {
                return a + b;
            }
        });

        //如果輸出檔案存在的話需要刪除掉
        File outputFile = new File("/Users/tangweiqun/wordcount");
        if (outputFile.exists()) {
            File[] files = outputFile.listFiles();
            for(File file: files) {
                file.delete();
            }
            outputFile.delete();
        }

        wordCountRDD.saveAsTextFile("file:///Users/tangweiqun/wordcount");

        System.out.println(wordCountRDD.collect());
    }
}
 

2、java8 lambda實現的wordcount程式碼

public class WordCount {

    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("appName").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);

        //JavaPairRDD<LongWritable, Text> inputRDD = sc.hadoopFile("hdfs://master:9999/user/word.txt",
        //        TextInputFormat.class, LongWritable.class, Text.class);

        JavaRDD<String> inputRDD = sc.textFile("file:///Users/tangweiqun/test.txt");

        JavaRDD<String> wordsRDD = inputRDD.flatMap(input -> Arrays.asList(input.split(" ")).iterator());

        JavaPairRDD<String, Integer> keyValueWordsRDD
                = wordsRDD.mapToPair(word -> new Tuple2<String, Integer>(word, 1));

        JavaPairRDD<String, Integer> wordCountRDD = keyValueWordsRDD.reduceByKey((a, b) -> a + b);

        //如果輸出檔案存在的話需要刪除掉
        File outputFile = new File("/Users/tangweiqun/wordcount");
        if (outputFile.exists()) {
            File[] files = outputFile.listFiles();
            for(File file: files) {
                file.delete();
            }
            outputFile.delete();
        }

        wordCountRDD.saveAsTextFile("file:///Users/tangweiqun/wordcount");

        System.out.println(wordCountRDD.collect());
    }
}
 

從上面可以看出，lambda的實現更加簡潔，也可以看出一個lambda函式表示式就是一個java介面。

JavaPairRDD<String, Integer> javaPairRDD =
        sc.parallelizePairs(Arrays.asList(new Tuple2("coffee", 1), new Tuple2("coffee", 2),
                new Tuple2("panda", 3), new Tuple2("coffee", 9)), 2);
 
//當在一個分割槽中遇到新的key的時候，對這個key對應的value應用這個函式
Function<Integer, Tuple2<Integer, Integer>> createCombiner = new Function<Integer, Tuple2<Integer, Integer>>() {
    @Override
    public Tuple2<Integer, Integer> call(Integer value) throws Exception {
        return new Tuple2<>(value, 1);
    }
};
//當在一個分割槽中遇到已經應用過上面createCombiner函式的key的時候，對這個key對應的value應用這個函式
Function2<Tuple2<Integer, Integer>, Integer, Tuple2<Integer, Integer>> mergeValue =
        new Function2<Tuple2<Integer, Integer>, Integer, Tuple2<Integer, Integer>>() {
            @Override
            public Tuple2<Integer, Integer> call(Tuple2<Integer, Integer> acc, Integer value) throws Exception {
                return new Tuple2<>(acc._1() + value, acc._2() + 1);
            }
        };
//當需要對不同分割槽的資料進行聚合的時候應用這個函式
Function2<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> mergeCombiners =
        new Function2<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>>() {
            @Override
            public Tuple2<Integer, Integer> call(Tuple2<Integer, Integer> acc1, Tuple2<Integer, Integer> acc2) throws Exception {
                return new Tuple2<>(acc1._1() + acc2._1(), acc1._2() + acc2._2());
            }
        };
 
JavaPairRDD<String, Tuple2<Integer, Integer>> combineByKeyRDD =
        javaPairRDD.combineByKey(createCombiner, mergeValue, mergeCombiners);
//結果：[(coffee,(12,3)), (panda,(3,1))]
System.out.println("combineByKeyRDD = " + combineByKeyRDD.collect());
 

可以寫成如下的lambda實現的combineByKey:

JavaPairRDD<String, Integer> javaPairRDD =
        sc.parallelizePairs(Arrays.asList(new Tuple2("coffee", 1), new Tuple2("coffee", 2),
                new Tuple2("panda", 3), new Tuple2("coffee", 9)), 2);
//當在一個分割槽中遇到新的key的時候，對這個key對應的value應用這個函式
Function<Integer, Tuple2<Integer, Integer>> createCombiner = value -> new Tuple2<>(value, 1);
//當在一個分割槽中遇到已經應用過上面createCombiner函式的key的時候，對這個key對應的value應用這個函式
Function2<Tuple2<Integer, Integer>, Integer, Tuple2<Integer, Integer>> mergeValue = (acc, value) ->new Tuple2<>(acc._1() + value, acc._2() + 1);
//當需要對不同分割槽的資料進行聚合的時候應用這個函式
Function2<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> mergeCombiners = (acc1, acc2) -> new Tuple2<>(acc1._1() + acc2._1(), acc1._2() + acc2._2());

JavaPairRDD<String, Tuple2<Integer, Integer>> combineByKeyRDD =
        javaPairRDD.combineByKey(createCombiner, mergeValue, mergeCombiners);
//結果：[(coffee,(12,3)), (panda,(3,1))]
System.out.println("combineByKeyRDD = " + combineByKeyRDD.collect());