improve sort examples

Author: James Lee

src/main/java/com/sparkTutorial/pairRdd/sort/sortbykey/AverageHousePriceSolution.java renamed to src/main/java/com/sparkTutorial/pairRdd/sort/AverageHousePriceSolution.java

@@ -1,4 +1,4 @@
-package com.sparkTutorial.pairRdd.sort.sortbykey;
+package com.sparkTutorial.pairRdd.sort;
 
 
 import com.sparkTutorial.pairRdd.aggregation.reducebykey.housePrice.AvgCount;
@@ -8,38 +8,26 @@
 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.Function2;
-import org.apache.spark.api.java.function.PairFunction;
 import scala.Tuple2;
 
-import java.util.Map;
-
 public class AverageHousePriceSolution {
 
     public static void main(String[] args) throws Exception {
-
         Logger.getLogger("org").setLevel(Level.ERROR);
         SparkConf conf = new SparkConf().setAppName("wordCounts").setMaster("local[3]");
         JavaSparkContext sc = new JavaSparkContext(conf);
 
         JavaRDD<String> lines = sc.textFile("in/RealEstate.csv");
-
         JavaRDD<String> cleanedLines = lines.filter(line -> !line.contains("Bedrooms"));
 
-        JavaPairRDD<String, AvgCount> housePricePairRdd = cleanedLines.mapToPair(
-                (PairFunction<String, String, AvgCount>) line ->
-                new Tuple2<>(line.split(",")[3],
-                             new AvgCount(1, Double.parseDouble(line.split(",")[2]))));
-
-        JavaPairRDD<String, AvgCount> housePriceTotal = housePricePairRdd.reduceByKey(
-                (Function2<AvgCount, AvgCount, AvgCount>) (x, y) ->
-                new AvgCount(x.getCount() + y.getCount(), x.getTotal() + y.getTotal()));
-
+        JavaPairRDD<Integer, AvgCount> housePricePairRdd = cleanedLines.mapToPair(
+                line -> new Tuple2<>(Integer.valueOf(line.split(",")[3]),
+                        new AvgCount(1, Double.parseDouble(line.split(",")[2]))));
 
-        JavaPairRDD<Integer, Double> housePriceAvg = housePriceTotal.mapToPair(
-                (PairFunction<Tuple2<String, AvgCount>, Integer, Double>) total ->
-                 new Tuple2<>(Integer.valueOf(total._1()), total._2().getTotal()/total._2().getCount()));
+        JavaPairRDD<Integer, AvgCount> housePriceTotal = housePricePairRdd.reduceByKey(
+                (x, y) -> new AvgCount(x.getCount() + y.getCount(), x.getTotal() + y.getTotal()));
 
+        JavaPairRDD<Integer, Double> housePriceAvg = housePriceTotal.mapValues(avgCount -> avgCount.getTotal()/avgCount.getCount());
 
         JavaPairRDD<Integer, Double> sortedHousePriceAvg = housePriceAvg.sortByKey();
 

src/main/java/com/sparkTutorial/pairRdd/sort/sortbykey/SortedWorldCountProblem.java renamed to src/main/java/com/sparkTutorial/pairRdd/sort/SortedWorldCountProblem.java

@@ -1,4 +1,4 @@
-package com.sparkTutorial.pairRdd.sort.sortbykey;
+package com.sparkTutorial.pairRdd.sort;
 
 
 public class SortedWorldCountProblem {

src/main/java/com/sparkTutorial/pairRdd/sort/sortbykey/SortedWorldCountSolution.java renamed to src/main/java/com/sparkTutorial/pairRdd/sort/SortedWorldCountSolution.java

@@ -1,4 +1,4 @@
-package com.sparkTutorial.pairRdd.sort.sortbykey;
+package com.sparkTutorial.pairRdd.sort;
 
 
 import org.apache.log4j.Level;
@@ -7,8 +7,6 @@
 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.Function2;
-import org.apache.spark.api.java.function.PairFunction;
 import scala.Tuple2;
 
 import java.util.Arrays;
@@ -24,24 +22,18 @@ public static void main(String[] args) throws Exception {
         JavaRDD<String> lines = sc.textFile("in/word_count.text");
         JavaRDD<String> wordRdd = lines.flatMap(line -> Arrays.asList(line.split(" ")).iterator());
 
-        JavaPairRDD<String, Integer> wordPairRdd = wordRdd.mapToPair(
-                (PairFunction<String, String, Integer>) word -> new Tuple2<>(word, 1));
-
-        JavaPairRDD<String, Integer> wordToCountPairs = wordPairRdd.reduceByKey((Function2<Integer, Integer, Integer>) (x, y) -> x + y);
-
-        JavaPairRDD<Integer, String> countToWordParis = wordToCountPairs.mapToPair(
-                (PairFunction<Tuple2<String, Integer>, Integer, String>) wordToCount -> new Tuple2<>(wordToCount._2(),
-                                                                                                     wordToCount._1()));
+        JavaPairRDD<String, Integer> wordPairRdd = wordRdd.mapToPair(word -> new Tuple2<>(word, 1));
+        JavaPairRDD<String, Integer> wordToCountPairs = wordPairRdd.reduceByKey((x, y) -> x + y);
 
+        JavaPairRDD<Integer, String> countToWordParis = wordToCountPairs.mapToPair(wordToCount -> new Tuple2<>(wordToCount._2(),
+                                                                                                               wordToCount._1()));
         JavaPairRDD<Integer, String> sortedCountToWordParis = countToWordParis.sortByKey(false);
 
         JavaPairRDD<String, Integer> sortedWordToCountPairs = sortedCountToWordParis
-                .mapToPair((PairFunction<Tuple2<Integer, String>, String, Integer>) countToWord -> new Tuple2<>(countToWord._2(),
-                                                                                                                countToWord._1()));
+                .mapToPair(countToWord -> new Tuple2<>(countToWord._2(), countToWord._1()));
 
         for (Tuple2<String, Integer> wordToCount : sortedWordToCountPairs.collect()) {
             System.out.println(wordToCount._1() + " : " + wordToCount._2());
-
         }
     }
 }