improve AirportsByCountrySolution and GroupByKeyVsReduceByKey

Author: James Lee

src/main/java/com/sparkTutorial/pairRdd/groupbykey/AirportsByCountrySolution.java

@@ -1,17 +1,15 @@
 package com.sparkTutorial.pairRdd.groupbykey;
 
-import com.google.common.collect.Iterables;
 import com.sparkTutorial.rdd.commons.Utils;
 import org.apache.log4j.Level;
 import org.apache.log4j.Logger;
 import org.apache.spark.SparkConf;
 import org.apache.spark.api.java.JavaPairRDD;
 import org.apache.spark.api.java.JavaRDD;
 import org.apache.spark.api.java.JavaSparkContext;
-import org.apache.spark.api.java.function.PairFunction;
 import scala.Tuple2;
 
-import java.util.Arrays;
+import java.util.Map;
 
 public class AirportsByCountrySolution {
 
@@ -23,18 +21,13 @@ public static void main(String[] args) throws Exception {
         JavaRDD<String> lines = sc.textFile("in/airports.text");
 
         JavaPairRDD<String, String> CountryAndAirportNameAndPair =
-                lines.mapToPair((PairFunction<String, String, String>) airport ->
-                        new Tuple2<>(airport.split(Utils.COMMA_DELIMITER)[3],
-                                     airport.split(Utils.COMMA_DELIMITER)[1]));
+                lines.mapToPair( airport -> new Tuple2<>(airport.split(Utils.COMMA_DELIMITER)[3],
+                                                         airport.split(Utils.COMMA_DELIMITER)[1]));
 
         JavaPairRDD<String, Iterable<String>> AirportsByCountry = CountryAndAirportNameAndPair.groupByKey();
 
-        for (Tuple2<String, Iterable<String>> airports : AirportsByCountry.collect()) {
-            System.out.println(airports._1() + " : " + iterableToString(airports._2()));
+        for (Map.Entry<String, Iterable<String>> airports : AirportsByCountry.collectAsMap().entrySet()) {
+            System.out.println(airports.getKey() + " : " + airports.getValue());
         }
     }
-
-    private static String iterableToString(Iterable<String> iterable) {
-        return Arrays.toString(Iterables.toArray(iterable, String.class));
-    }
 }

src/main/java/com/sparkTutorial/pairRdd/groupbykey/GroupByKeyVsReduceByKey.java

@@ -1,5 +1,8 @@
 package com.sparkTutorial.pairRdd.groupbykey;
 
+import com.google.common.collect.Iterables;
+import org.apache.log4j.Level;
+import org.apache.log4j.Logger;
 import org.apache.spark.SparkConf;
 import org.apache.spark.api.java.JavaPairRDD;
 import org.apache.spark.api.java.JavaSparkContext;
@@ -11,26 +14,19 @@
 public class GroupByKeyVsReduceByKey {
 
     public static void main(String[] args) throws Exception {
-
+        Logger.getLogger("org").setLevel(Level.ERROR);
         SparkConf conf = new SparkConf().setAppName("GroupByKeyVsReduceByKey").setMaster("local[*]");
         JavaSparkContext sc = new JavaSparkContext(conf);
 
         List<String> words = Arrays.asList("one", "two", "two", "three", "three", "three");
-
         JavaPairRDD<String, Integer> wordsPairRdd = sc.parallelize(words).mapToPair(word -> new Tuple2<>(word, 1));
 
-        List<Tuple2<String, Integer>> wordCountsWithReduce = wordsPairRdd.reduceByKey((x, y) -> x + y).collect();
-
-        List<Tuple2<String, Integer>> wordCountsWithGroup = wordsPairRdd.groupByKey()
-                .mapToPair(word -> new Tuple2<>(word._1(), getSum(word._2()))).collect();
-    }
+        List<Tuple2<String, Integer>> wordCountsWithReduceByKey = wordsPairRdd.reduceByKey((x, y) -> x + y).collect();
+        System.out.println("wordCountsWithReduceByKey: " + wordCountsWithReduceByKey);
 
-    private static int getSum(Iterable<Integer> integers) {
-        int sum = 0;
-        for (Integer integer : integers) {
-            sum = + integer;
-        }
-        return sum;
+        List<Tuple2<String, Integer>> wordCountsWithGroupByKey = wordsPairRdd.groupByKey()
+                .mapValues(intIterable -> Iterables.size(intIterable)).collect();
+        System.out.println("wordCountsWithGroupByKey: " + wordCountsWithGroupByKey);
     }
 }