package com.naughtyzombie.dataanalysis.sparkexample;

import org.apache.spark.SparkConf;
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.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.util.Arrays;

public class WordCount {
    private static final FlatMapFunction<String, String> WORDS_EXTRACTOR =
            s -> Arrays.asList(s.split(" "));

    private static final PairFunction<String, String, Integer> WORDS_MAPPER =
            s -> new Tuple2<String, Integer>(s, 1);

    private static final Function2<Integer, Integer, Integer> WORDS_REDUCER =
            (a, b) -> a + b;

    public static void main(String[] args) {
        if (args.length < 1) {
            System.err.println("Please provide the input file full path as argument");
            System.exit(0);
        }

        SparkConf conf = new SparkConf().setAppName("org.sparkexample.WordCount").setMaster("local");
        JavaSparkContext context = new JavaSparkContext(conf);

        JavaRDD<String> file = context.textFile(args[0]);
        JavaRDD<String> words = file.flatMap(WORDS_EXTRACTOR);
        JavaPairRDD<String, Integer> pairs = words.mapToPair(WORDS_MAPPER);
        JavaPairRDD<String, Integer> counter = pairs.reduceByKey(WORDS_REDUCER);

        counter.saveAsTextFile(args[1]);
    }
}
@@ -462,18 +462,16 @@
 Count %7B%0A
-  
   privat
@@ -541,26 +541,156 @@
 =%0A      
-      s -%3E
+new FlatMapFunction%3CString, String%3E() %7B%0A        @Override%0A        public Iterable%3CString%3E call(String s) throws Exception %7B%0A          return
  Arrays.
@@ -711,19 +711,36 @@
 (%22 %22));%0A
-%0A
   
+      %7D%0A      %7D;%0A%0A
   privat
@@ -817,18 +817,161 @@
     
-      s -%3E
+new PairFunction%3CString, String, Integer%3E() %7B%0A        @Override%0A        public Tuple2%3CString, Integer%3E call(String s) throws Exception %7B%0A          return
  new
@@ -1002,19 +1002,36 @@
 (s, 1);%0A
-%0A
   
+      %7D%0A      %7D;%0A%0A
   privat
@@ -1108,34 +1108,184 @@
     
-      (a, b) -%3E a + b;%0A%0A  
+new Function2%3CInteger, Integer, Integer%3E() %7B%0A        @Override%0A        public Integer call(Integer a, Integer b) throws Exception %7B%0A          return a + b;%0A        %7D%0A      %7D;%0A%0A
   pu
@@ -1327,20 +1327,16 @@
 ) %7B%0A    
-  
-  
 if (args
@@ -1350,22 +1350,16 @@
  %3C 1) %7B%0A
-      
       Sy
@@ -1431,22 +1431,16 @@
 ment%22);%0A
-      
       Sy
@@ -1461,19 +1461,11 @@
     
-  
-  
 %7D%0A%0A
-    
     
@@ -1558,28 +1558,24 @@
 ocal%22);%0A    
-  
-  
 JavaSparkCon
@@ -1614,28 +1614,24 @@
 ext(conf);%0A%0A
-    
     JavaRDD%3C
@@ -1672,28 +1672,24 @@
 gs%5B0%5D);%0A    
-  
-  
 JavaRDD%3CStri
@@ -1727,28 +1727,24 @@
 EXTRACTOR);%0A
-    
     JavaPair
@@ -1807,20 +1807,16 @@
 R);%0A    
-  
-  
 JavaPair
@@ -1881,20 +1881,16 @@
 UCER);%0A%0A
-    
     coun
@@ -1918,15 +1918,13 @@
 gs%5B1%5D);%0A
-  
   %7D%0A%7D

remove lambdas