improve AverageHousePriceSolution

Author: James Lee

src/main/java/com/sparkTutorial/pairRdd/aggregation/combinebykey/AverageHousePriceSolution.java

@@ -9,7 +9,6 @@
 import org.apache.spark.api.java.JavaSparkContext;
 import org.apache.spark.api.java.function.Function;
 import org.apache.spark.api.java.function.Function2;
-import org.apache.spark.api.java.function.PairFunction;
 import scala.Tuple2;
 
 import java.util.Map;
@@ -23,33 +22,26 @@ public static void main(String[] args) throws Exception {
         JavaSparkContext sc = new JavaSparkContext(conf);
 
         JavaRDD<String> lines = sc.textFile("in/RealEstate.csv");
-
         JavaRDD<String> cleanedLines = lines.filter(line -> !line.contains("Bedrooms"));
 
         JavaPairRDD<String, Double> housePricePairRdd = cleanedLines.mapToPair(
-                (PairFunction<String, String, Double>) line ->
-                new Tuple2<>(line.split(",")[3],
-                             Double.parseDouble(line.split(",")[2])));
+                 line -> new Tuple2<>(line.split(",")[3],
+                                      Double.parseDouble(line.split(",")[2])));
 
         JavaPairRDD<String, AvgCount> housePriceTotal= housePricePairRdd.combineByKey(createCombiner, mergeValue, mergeCombiners);
 
-        JavaPairRDD<String, Double> housePriceAvg = housePriceTotal.mapToPair(
-                (PairFunction<Tuple2<String, AvgCount>, String, Double>) total ->
-                 new Tuple2<>(total._1(), total._2().getTotal()/total._2().getCount()));
+        JavaPairRDD<String, Double> housePriceAvg = housePriceTotal.mapValues(avgCount -> avgCount.getTotal()/avgCount.getCount());
 
         for (Map.Entry<String, Double> housePriceAvgPair : housePriceAvg.collectAsMap().entrySet()) {
             System.out.println(housePriceAvgPair.getKey() + " : " + housePriceAvgPair.getValue());
-
         }
     }
 
-    static Function<Double, AvgCount> createCombiner = (Function<Double, AvgCount>) x -> new AvgCount(1, x);
-
-    static Function2<AvgCount, Double, AvgCount> mergeValue
-            = (Function2<AvgCount, Double, AvgCount>) (avgCount, x) -> new AvgCount(avgCount.getCount() + 1,
-                                                                                    avgCount.getTotal() + x);
-    static Function2<AvgCount, AvgCount, AvgCount> mergeCombiners
-            = (Function2<AvgCount, AvgCount, AvgCount>) (avgCountA, avgCountB) -> new AvgCount(avgCountA.getCount() + avgCountB.getCount(),
-                                                                                               avgCountA.getTotal() + avgCountB.getTotal());
+    static Function<Double, AvgCount> createCombiner = x -> new AvgCount(1, x);
 
+    static Function2<AvgCount, Double, AvgCount> mergeValue = (avgCount, x) -> new AvgCount(avgCount.getCount() + 1,
+                                                                                             avgCount.getTotal() + x);
+    static Function2<AvgCount, AvgCount, AvgCount> mergeCombiners =
+            (avgCountA, avgCountB) -> new AvgCount(avgCountA.getCount() + avgCountB.getCount(),
+                                                   avgCountA.getTotal() + avgCountB.getTotal());
 }