package ch.bfh.cas.bgd.ta;

import java.util.Arrays;
import java.util.Iterator;

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;

public class WordCount {

	@SuppressWarnings("serial")
	public static void main(String[] args) {
		if (args.length == 0) {
			System.out.println("Usage: WordCount <file>");
			System.exit(0);
		}

		// Define a configuration to use to interact with Spark
		SparkConf conf = new SparkConf().setMaster("spark://bigdata-VirtualBox:7077").setAppName("Word Count App"); // TODO may need to change the master URL

		// Create a Java version of the Spark Context from the configuration
		JavaSparkContext sc = new JavaSparkContext(conf);

		// 1. prepare the input data
		// 1.1 load the input data, which is a text file read from the command line
		JavaRDD<String> input = sc.textFile(args[0]);

		// 1.2 read all words from the textfile
		JavaRDD<String> words = input.flatMap(new FlatMapFunction<String, String>() {
			public Iterator<String> call(String s) {
				return Arrays.asList(s.split(" ")).iterator(); // very crude tokenization via space character
			}
		});

		// 2. map - transform the collection of words into pairs of <word, 1>
		JavaPairRDD<String, Integer> counts = words.mapToPair(new PairFunction<String, String, Integer>() {
			public Tuple2<String, Integer> call(String s) {
				return new Tuple2<String, Integer>(s, 1);
			}
		});

		// 3. reduce - count the words
		JavaPairRDD<String, Integer> reducedCounts = counts.reduceByKey(new Function2<Integer, Integer, Integer>() {
			public Integer call(Integer x, Integer y) {
				return x + y;
			}
		});

		// Save the word count to a text file
		reducedCounts.saveAsTextFile("output");
		
		// cleanup
		sc.close();
	}
}

@@ -42,41 +42,21 @@
 til.
-Arrays;%0Aimport java.util.Iterator
+regex.Pattern
 ;%0A%0Ai
@@ -231,176 +231,8 @@
 ext;
-%0Aimport org.apache.spark.api.java.function.FlatMapFunction;%0Aimport org.apache.spark.api.java.function.Function2;%0Aimport org.apache.spark.api.java.function.PairFunction;
 %0A%0Aim
@@ -281,37 +281,8 @@
  %7B%0A%0A
-%09@SuppressWarnings(%22serial%22)%0A
 %09pub
@@ -319,16 +319,17 @@
 args) %7B%0A
+%0A
 %09%09if (ar
@@ -478,16 +478,61 @@
 h Spark%0A
+%09%09// TODO: may need to change the master URL%0A
 %09%09SparkC
@@ -636,50 +636,8 @@
 p%22);
- // TODO may need to change the master URL
 %0A%0A%09%09
@@ -707,16 +707,21 @@
 ation%0A%09%09
+try (
 JavaSpar
@@ -760,19 +760,22 @@
 xt(conf)
-;
+) %7B
 %0A%0A
+%09
 %09%09// 1. 
@@ -797,16 +797,17 @@
 ut data%0A
+%09
 %09%09// 1.1
@@ -874,24 +874,25 @@
 mand line%0A%09%09
+%09
 JavaRDD%3CStri
@@ -928,16 +928,17 @@
 0%5D);%0A%0A%09%09
+%09
 // 1.2 r
@@ -968,247 +968,146 @@
 text
+ 
 file%0A%09%09
-JavaRDD%3CString%3E words = input.flatMap(new FlatMapFunction%3CString, String%3E() %7B%0A%09%09%09public Iterator%3CString%3E call(String s) %7B%0A%09%09%09%09return Arrays.asList(s.split(%22 %22)).iterator(); // very crude tokenization via space character%0A%09%09%09%7D%0A%09%09%7D
+%09Pattern pattern = Pattern.compile(%22%5C%5CW+%22);%0A%09%09%09JavaRDD%3CString%3E words = input.flatMap(s -%3E pattern.splitAsStream(s).iterator()
 );%0A%0A
+%09
 %09%09//
@@ -1176,16 +1176,17 @@
 d, 1%3E%0A%09%09
+%09
 JavaPair
@@ -1235,115 +1235,12 @@
 air(
-new PairFunction%3CString, String, Integer%3E() %7B%0A%09%09%09public Tuple2%3CString, Integer%3E call(String s) %7B%0A%09%09%09%09return
+s -%3E
  new
@@ -1269,30 +1269,21 @@
 r%3E(s, 1)
-;%0A%09%09%09%7D%0A%09%09%7D
 );%0A%0A
+%09
 %09%09// 3. 
@@ -1307,16 +1307,17 @@
 e words%0A
+%09
 %09%09JavaPa
@@ -1378,130 +1378,28 @@
 Key(
-new Function2%3CInteger, Integer, Integer%3E() %7B%0A%09%09%09public Integer call(Integer x, Integer y) %7B%0A%09%09%09%09return
+(x, y) -%3E
  x + y
+)
 ;%0A
-%09%09%09%7D%0A%09%09%7D);%0A
 %0A
+%09
 %09%09//
@@ -1436,16 +1436,17 @@
  file%0A%09%09
+%09
 reducedC
@@ -1451,16 +1451,28 @@
 dCounts.
+coalesce(1).
 saveAsTe
@@ -1495,35 +1495,9 @@
 ;%0A%09%09
-%0A%09%09// cleanup%0A%09%09sc.close();
+%7D
 %0A%09%7D%0A

improve wordcount