/**
 * Illustrates using counters and broadcast variables for chapter 6
 */
package com.oreilly.learningsparkexamples.java;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;


import org.apache.commons.lang.StringUtils;

import org.apache.spark.Accumulator;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;

public class ChapterSixExample {
  public static void main(String[] args) throws Exception {

		if (args.length != 3) {
      throw new Exception("Usage AccumulatorExample sparkMaster inputFile outDirectory");
		}

    JavaSparkContext sc = new JavaSparkContext(
      args[0], "ChapterSixExample", System.getenv("SPARK_HOME"), System.getenv("JARS"));
    // Create Accumulators initialized at 0
    final Accumulator<Integer> blankLines = sc.accumulator(0);
    JavaRDD<String> rdd = sc.textFile(args[1]);
    JavaRDD<String> callSigns = rdd.flatMap(
      new FlatMapFunction<String, String>() { public Iterable<String> call(String line) {
          if (line.equals("")) {
            blankLines.add(1);
          }
          return Arrays.asList(line.split(" "));
        }});
    // Force evaluation so the counters are populated
    callSigns.count();
    System.out.println("Blank lines: "+ blankLines.value());
  }
}

@@ -216,16 +216,41 @@
 .Entry;%0A
+import java.util.regex.*;
 %0A%0Aimport
@@ -522,16 +522,68 @@
 unction;
+%0Aimport org.apache.spark.api.java.function.Function;
 %0A%0Apublic
@@ -789,16 +789,114 @@
 y%22);%0A%09%09%7D
+%0A    String sparkMaster = args%5B0%5D;%0A    String inputFile = args%5B1%5D;%0A    String outputDir = args%5B2%5D;
 %0A%0A    Ja
@@ -943,23 +943,27 @@
 (%0A      
-args%5B0%5D
+sparkMaster
 , %22Chapt
@@ -1175,23 +1175,25 @@
 extFile(
-args%5B1%5D
+inputFile
 );%0A    J
@@ -1460,16 +1460,770 @@
    %7D%7D);%0A
+    callSigns.saveAsTextFile(outputDir + %22/callsigns%22);%0A    System.out.println(%22Blank lines: %22+ blankLines.value());%0A    // Start validating the call signs%0A    final Accumulator%3CInteger%3E validSignCount = sc.accumulator(0);%0A    final Accumulator%3CInteger%3E invalidSignCount = sc.accumulator(0);%0A    JavaRDD%3CString%3E validCallSigns = callSigns.filter(%0A      new Function%3CString, Boolean%3E()%7B public Boolean call(String callSign) %7B%0A          Pattern p = Pattern.compile(%22%5C%5CA%5C%5Cd%5C%5Cp%7BAlpha%7D%7B1,2%7D%5C%5Cd%7B1,4%7D%5C%5C%7BAlpha%7D%7B1,3%7D%5C%5CZ%22);%0A          Matcher m = p.matcher(callSign);%0A          boolean b = m.matches();%0A          if (b) %7B%0A            validSignCount.add(1);%0A          %7D else %7B%0A            invalidSignCount.add(1);%0A          %7D%0A          return b;%0A        %7D%0A      %7D);%0A
     // F
@@ -2268,25 +2268,30 @@
 pulated%0A    
-c
+validC
 allSigns.cou
@@ -2304,64 +2304,287 @@
     
-System.out.println(%22Blank lines:
+if (invalidSignCount.value() %3C 0.1 * validSignCount.value()) %7B%0A      validCallSigns.saveAsTextFile(outputDir + %22/validatedSigns%22);%0A    %7D else %7B%0A      System.out.println(%22Too many errors %22 + invalidSignCount.value() + %22 for
  %22
+ 
 + 
-blankLines.value());
+validSignCount.value());%0A      System.exit(1);%0A    %7D
 %0A  %7D

Update the chapter six example to validate the call signs