Notes

import scala.Tuple2;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.OutputCollector;
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.*;

import java.io.IOException;
import java.io.NotSerializableException;
import java.io.Serializable;
import java.text.NumberFormat;
import java.util.*;
import java.util.regex.Pattern;

public final class wordcount implements Serializable{
private static final Pattern SPACE = Pattern.compile(" ");
private final static IntWritable one = new IntWritable(1);

//public static List<Tuple2<Text,IntWritable>> mapper(Text value, OutputCollector<Text, IntWritable> output) throws IOException
public static List<Tuple2<Text,IntWritable>> mapper(Text value, OutputCollector<Text, IntWritable> output) throws IOException
{
//NotSerializableException notSerializable = new NotSerializableException();
String line = value.toString();
StringTokenizer tokenizer = new StringTokenizer(line);
Text word = new Text();
while (tokenizer.hasMoreTokens()){
word = new Text();
word.set(tokenizer.nextToken());
//sending to output collector which inturn passes the same to reducer
output.collect(word, one);
}
return ((outputcollector<Text, IntWritable>)output).getList();
}

public static Tuple2<Text,IntWritable> reduce(Text key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output) throws IOException
{
int sum = 0;
while (values.hasNext())
{
sum += values.next().get();
}
output.collect(key, new IntWritable(sum));
return ((outputcollector<Text, IntWritable>)output).getList().get(0);
}

// public static String memusage(){
// Runtime runtime = Runtime.getRuntime();
// NumberFormat format = NumberFormat.getInstance();
// StringBuilder sb = new StringBuilder();
// long maxMemory = runtime.maxMemory();
// long allocatedMemory = runtime.totalMemory();
// long freeMemory = runtime.freeMemory();
//
// sb.append("free memory: " + format.format(freeMemory / 1024) + "<br/>");
// sb.append("allocated memory: " + format.format(allocatedMemory / 1024) + "<br/>");
// sb.append("max memory: " + format.format(maxMemory / 1024) + "<br/>");
// sb.append("total free memory: " + format.format((freeMemory + (maxMemory - allocatedMemory)) / 1024) + "<br/>");
// return sb.toString();
// }



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

if (args.length < 1) {
System.err.println("Use arguments to give input filesnnn");
System.exit(1);
}

SparkConf sparkConf = new SparkConf().setMaster("spark://IMPETUS-I0203:7077").setAppName("wordcount").set("spark.executor.memory", "2g");
JavaSparkContext ctx = new JavaSparkContext(sparkConf);
JavaRDD<String> lines = ctx.textFile(args[0], 1);
//List<Tuple2<Text,IntWritable>> list;
JavaRDD<String> words = lines.flatMap(s -> Arrays.asList(SPACE.split(s)));

//wordcount w = new wordcount();
JavaPairRDD<Text,IntWritable> ones = words.flatMapToPair(s -> {
OutputCollector<Text,IntWritable> output = new outputcollector<Text,IntWritable>() ;
//memusage();
return wordcount.mapper(new Text(s),output);
}
);

//JavaPairRDD<Text,IntWritable> ones = words.mapToPair(s -> new Tuple2<Text,IntWritable>(new Text(s),new IntWritable(1)));
//JavaPairRDD<String,Integer> ones=words.mapToPair(s->new Tuple2<String,Integer>(s,1));
JavaPairRDD<String,Integer> one = ones.mapToPair(a -> new Tuple2<String,Integer>(a._1.toString(),a._2.get() ));
//JavaPairRDD<String,Integer> counts = one.reduceByKey((a,b)->a+b);
JavaPairRDD<String,Iterable<Integer>> grouped = one.groupByKey();

/* JavaPairRDD<Text,Iterable<IntWritable>> grouped_new = grouped.mapToPair(t-> {
List<IntWritable> list = new ArrayList<IntWritable>();
//t._2()
for (Integer a : t._2)
{
list.add(new IntWritable(a.intValue()));
}
return new Tuple2<Text,Iterable<IntWritable>>(new Text(t._1),list);
});*/


//JavaPairRDD<Text,IntWritable> counts = ones.reduceByKey((i1,i2)-> new IntWritable(i1.get()+i2.get()));
JavaPairRDD<Text,IntWritable> counts = grouped.mapToPair(t->{
List<IntWritable> list = new ArrayList<IntWritable>();
for (Integer a : t._2)
{
list.add(new IntWritable(a.intValue()));
}
OutputCollector<Text,IntWritable> output = new outputcollector<Text,IntWritable>() ;
return wordcount.reduce(new Text(t._1), list.iterator(), output);
});


counts.saveAsTextFile(args[1]);
ctx.stop();
}
}