import java.io.IOException;
import java.lang.InterruptedException;
import java.util.StringTokenizer;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class DeviceTraffic {

//public static String publisherId;
Configuration conf = new Configuration();
static class DeviceTrafficMapper
extends Mapper<Object, Text, Text, IntWritable> {
//private static final int MISSING = 9999;

public void map(Object key, Text value, Context context)
throws IOException, InterruptedException {
//Configuration mconf = new Configuration();
//publisherId = mconf.get("PublisherId");
String mpublisherId=conf.get("PublisherId");
String line = value.toString();
String[] splitValues = line.split("\t");
System.out.println("Split Value size is "+String.valueOf(splitValues.length));
if (splitValues!=null && splitValues.length>2)
{
String pubPaymentId=splitValues[24];
String device;
int impression;
System.out.println("Publisher id is "+mpublisherId);
System.out.println("Publisher Payment id is "+pubPaymentId);
if (pubPaymentId.equals(mpublisherId))
{
if (splitValues[33]!=null && !splitValues[33].equals("") && !splitValues[33].equals("ELI"))
{
device = splitValues[33];
//if (splitValues[20]!=null && !splitValues[20].equals("") && !splitValues[20].equals("Fraud") && !splitValues[20].equals("ELI"))
impression = Integer.valueOf(splitValues[20]);
//impression = 1;
//else
//impression = 0;
context.write(new Text(value), new IntWritable(impression));
}
}
else
{
context.write(new Text(mpublisherId), new IntWritable(1));
}
}
}
}
static class DeviceTrafficReducer
extends Reducer<Text, IntWritable, Text, IntWritable> {
public void reduce(Text key, Iterable<IntWritable> values,
Context context)
throws IOException, InterruptedException {
int sumValue = 0;
//BufferedWriter writer = new BufferedWriter(new FileWriter("/home/hadoop/ads_deliveries/output.txt"));
for (IntWritable value : values) {
//writer.write(String.valueOf(value)+"\t");
sumValue +=  value.get();
}
context.write(key, new IntWritable(sumValue));
}
}

public static void main(String[] args) throws Exception {
if (args.length != 3) {
System.err.println("Usage: PublisherUrl <publihser id> <input path> <output path>");
System.exit(-1);
}
conf.setIfUnset("PublisherId",args[0]);
//publisherId=args[0];
System.out.println("Publisher ID in argument is "+args[0]);
DeviceTraffic.setpubId(args[0]);
Job job = new Job(conf);
job.setJarByClass(DeviceTraffic.class);
FileInputFormat.addInputPath(job, new Path(args[1]));
FileOutputFormat.setOutputPath(job, new Path(args[2]));
job.setMapperClass(DeviceTrafficMapper.class);
job.setReducerClass(DeviceTrafficReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}