add StackOverFlowSurveyFollowUp

James Lee · James Lee · commit 86eee4e41c15 · 2017-02-11T12:20:37.000Z
diff --git a/src/main/java/com/sparkTutorial/advanced/accumulator/StackOverFlowSurvey.java b/src/main/java/com/sparkTutorial/advanced/accumulator/StackOverFlowSurvey.java
@@ -4,6 +4,7 @@
 import org.apache.spark.SparkContext;
 import org.apache.spark.api.java.JavaRDD;
 import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.util.AccumulatorV2;
 import org.apache.spark.util.LongAccumulator;
 import scala.Option;
 
@@ -23,7 +24,6 @@ public static void main(String[] args) throws Exception {
         total.register(sparkContext, Option.apply("total"), false);
         missingSalaryMidPoint.register(sparkContext, Option.apply("missing salary middle point"), false);
 
-
         JavaRDD<String> responseRDD = javaSparkContext.textFile("in/2016-stack-overflow-survey-responses.csv");
 
         JavaRDD<String> responseFromCanada = responseRDD.filter(response -> {
diff --git a/src/main/java/com/sparkTutorial/advanced/accumulator/StackOverFlowSurveyFollowUp.java b/src/main/java/com/sparkTutorial/advanced/accumulator/StackOverFlowSurveyFollowUp.java
@@ -0,0 +1,51 @@
+package com.sparkTutorial.advanced.accumulator;
+
+import org.apache.spark.SparkConf;
+import org.apache.spark.SparkContext;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.util.LongAccumulator;
+import scala.Option;
+
+public class StackOverFlowSurveyFollowUp {
+
+    public static void main(String[] args) throws Exception {
+
+        SparkConf conf = new SparkConf().setAppName("StackOverFlowSurvey").setMaster("local[1]");
+
+        SparkContext sparkContext = new SparkContext(conf);
+
+        JavaSparkContext javaSparkContext = new JavaSparkContext(sparkContext);
+
+        final LongAccumulator total = new LongAccumulator();
+        final LongAccumulator missingSalaryMidPoint = new LongAccumulator();
+        final LongAccumulator processedBytes = new LongAccumulator();
+
+        total.register(sparkContext, Option.apply("total"), false);
+        missingSalaryMidPoint.register(sparkContext, Option.apply("missing salary middle point"), false);
+        processedBytes.register(sparkContext, Option.apply("Processed bytes"), true);
+
+        JavaRDD<String> responseRDD = javaSparkContext.textFile("in/2016-stack-overflow-survey-responses.csv");
+
+        JavaRDD<String> responseFromCanada = responseRDD.filter(response -> {
+
+            processedBytes.add(response.getBytes().length);
+
+            String[] splits = response.split(",", -1);
+
+            total.add(1);
+
+            if (splits[14].equals("")) {
+                missingSalaryMidPoint.add(1);
+            }
+
+            return splits[2].equals("Canada");
+
+        });
+
+        System.out.println("Count of responses from Canada: " + responseFromCanada.count());
+        System.out.println("Number of bytes processed: " + processedBytes.value());
+        System.out.println("Total count of responses: " + total.value());
+        System.out.println("Count of responses missing salary middle point: " + missingSalaryMidPoint.value());
+    }
+}
