improve accumulator and broadcast exmaples

Author: James Lee

src/main/java/com/sparkTutorial/advanced/accumulator/StackOverFlowSurvey.java

@@ -1,5 +1,8 @@
 package com.sparkTutorial.advanced.accumulator;
 
+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.SparkContext;
 import org.apache.spark.api.java.JavaRDD;
@@ -10,11 +13,10 @@
 public class StackOverFlowSurvey {
 
     public static void main(String[] args) throws Exception {
-
+        Logger.getLogger("org").setLevel(Level.ERROR);
         SparkConf conf = new SparkConf().setAppName("StackOverFlowSurvey").setMaster("local[1]");
 
         SparkContext sparkContext = new SparkContext(conf);
-
         JavaSparkContext javaSparkContext = new JavaSparkContext(sparkContext);
 
         final LongAccumulator total = new LongAccumulator();
@@ -26,11 +28,11 @@ public static void main(String[] args) throws Exception {
         JavaRDD<String> responseRDD = javaSparkContext.textFile("in/2016-stack-overflow-survey-responses.csv");
 
         JavaRDD<String> responseFromCanada = responseRDD.filter(response -> {
-            String[] splits = response.split(",", -1);
+            String[] splits = response.split(Utils.COMMA_DELIMITER, -1);
 
             total.add(1);
 
-            if (splits[14].equals("")) {
+            if (splits[14].isEmpty()) {
                 missingSalaryMidPoint.add(1);
             }
 

src/main/java/com/sparkTutorial/advanced/accumulator/StackOverFlowSurveyFollowUp.java

@@ -1,5 +1,8 @@
 package com.sparkTutorial.advanced.accumulator;
 
+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.SparkContext;
 import org.apache.spark.api.java.JavaRDD;
@@ -10,11 +13,9 @@
 public class StackOverFlowSurveyFollowUp {
 
     public static void main(String[] args) throws Exception {
-
+        Logger.getLogger("org").setLevel(Level.ERROR);
         SparkConf conf = new SparkConf().setAppName("StackOverFlowSurvey").setMaster("local[1]");
-
         SparkContext sparkContext = new SparkContext(conf);
-
         JavaSparkContext javaSparkContext = new JavaSparkContext(sparkContext);
 
         final LongAccumulator total = new LongAccumulator();
@@ -31,11 +32,11 @@ public static void main(String[] args) throws Exception {
 
             processedBytes.add(response.getBytes().length);
 
-            String[] splits = response.split(",", -1);
+            String[] splits = response.split(Utils.COMMA_DELIMITER, -1);
 
             total.add(1);
 
-            if (splits[14].equals("")) {
+            if (splits[14].isEmpty()) {
                 missingSalaryMidPoint.add(1);
             }
 

src/main/java/com/sparkTutorial/advanced/broadcast/UkMarketSpaces.java

@@ -1,5 +1,6 @@
 package com.sparkTutorial.advanced.broadcast;
 
+import com.sparkTutorial.rdd.commons.Utils;
 import org.apache.log4j.Level;
 import org.apache.log4j.Logger;
 import org.apache.spark.SparkConf;
@@ -14,19 +15,16 @@
 public class UkMarketSpaces {
 
     public static void main(String[] args) throws Exception {
-
         Logger.getLogger("org").setLevel(Level.ERROR);
-
         SparkConf conf = new SparkConf().setAppName("UkMarketSpaces").setMaster("local[1]");
-
         JavaSparkContext javaSparkContext = new JavaSparkContext(conf);
 
         final Broadcast<Map<String, String>> postCodeMap = javaSparkContext.broadcast(loadPostCodeMap());
 
         JavaRDD<String> marketsRdd = javaSparkContext.textFile("in/uk-market-spaces-identifiable-data.csv");
 
         JavaRDD<String> regions = marketsRdd
-                .filter(line -> !line.split(",", -1)[0].equals("Timestamp"))
+                .filter(line -> !line.split(Utils.COMMA_DELIMITER, -1)[0].equals("Timestamp"))
                 .map(line -> {
                     Optional<String> postPrefix = getPostPrefix(line);
                     if (postPrefix.isPresent() && postCodeMap.value().containsKey(postPrefix.get())) {
@@ -40,7 +38,7 @@ public static void main(String[] args) throws Exception {
     }
 
     private static Optional<String> getPostPrefix(String line) {
-        String[] splits = line.split(",", -1);
+        String[] splits = line.split(Utils.COMMA_DELIMITER, -1);
         String postcode = splits[4];
         if (postcode.isEmpty()) {
             return Optional.empty();
@@ -53,7 +51,7 @@ private static Map<String, String> loadPostCodeMap() throws FileNotFoundExceptio
         Map<String, String> postCodeMap = new HashMap<>();
         while (postCode.hasNextLine()) {
             String line = postCode.nextLine();
-            String[] splits = line.split(",", -1);
+            String[] splits = line.split(Utils.COMMA_DELIMITER, -1);
             postCodeMap.put(splits[0], splits[7]);
         }
         return postCodeMap;

src/main/java/com/sparkTutorial/advanced/broadcast/UkMarketSpacesWithoutBroadcaset.java

@@ -1,5 +1,8 @@
 package com.sparkTutorial.advanced.broadcast;
 
+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.JavaRDD;
 import org.apache.spark.api.java.JavaSparkContext;
@@ -11,17 +14,16 @@
 public class UkMarketSpacesWithoutBroadcaset {
 
     public static void main(String[] args) throws Exception {
-
+        Logger.getLogger("org").setLevel(Level.ERROR);
         SparkConf conf = new SparkConf().setAppName("UkMarketSpaces").setMaster("local[1]");
-
         JavaSparkContext javaSparkContext = new JavaSparkContext(conf);
 
         final Map<String, String> postCodeMap = loadPostCodeMap();
 
         JavaRDD<String> marketsRdd = javaSparkContext.textFile("in/uk-market-spaces-identifiable-data.csv");
 
         JavaRDD<String> regions = marketsRdd
-                .filter(line -> !line.split(",", -1)[0].equals("Timestamp"))
+                .filter(line -> !line.split(Utils.COMMA_DELIMITER, -1)[0].equals("Timestamp"))
                 .map(line -> {
                     List<String> postCodePrefixes = getPostPrefixes(line);
                     for (String  postCodePrefix: postCodePrefixes) {
@@ -37,7 +39,7 @@ public static void main(String[] args) throws Exception {
     }
 
     private static List<String> getPostPrefixes(String line) {
-        String[] splits = line.split(",", -1);
+        String[] splits = line.split(Utils.COMMA_DELIMITER, -1);
         String postcode = splits[4];
         String cleanedPostCode = postcode.replaceAll("\\s+", "");
         ArrayList<String> prefixes = new ArrayList<>();
@@ -52,7 +54,7 @@ private static Map<String, String> loadPostCodeMap() throws FileNotFoundExceptio
         Map<String, String> postCodeMap = new HashMap<>();
         while (postCode.hasNextLine()) {
             String line = postCode.nextLine();
-            String[] splits = line.split(",", -1);
+            String[] splits = line.split(Utils.COMMA_DELIMITER, -1);
             postCodeMap.put(splits[0], splits[7]);
         }
         return  postCodeMap;