ReadCSV.java

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

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
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.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class ReadCSV 
{
	public static class CountMapper extends Mapper<LongWritable, Text, Text, Text> {
		/*static enum SalesCounters {
			MISSING, INVALID
		};*/

		public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

			// Input string is split using ',' and stored in 'fields' array
			String fields[] = value.toString().split(",");
			// Value at 2th index is country. It is stored in 'country' variable
			String country = fields[2];

			// Value at 4th index is sales data. It is stored in 'sales' variable
			String sales = fields[3];

			if (country.length() == 0)
			{
				//context.getCounter(SalesCounters.MISSING).increment(1L);
				context.write(new Text(country), new Text("MISSING"));
			} 
			else if (!(Character.isDigit(sales.charAt(0))))
			{
				//context.getCounter(SalesCounters.INVALID).increment(1L);
				context.write(new Text(country), new Text("INVALID"));
			}
			else
			{
				context.write(new Text(country), new Text(sales));
			}
		}
	}
	public static class CountReducer extends Reducer<Text, Text, Text, Text> 
	{
		public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException 
		{
			String total="";
			for (Text details : values) 
			{
				total=total+" "+details.toString();
			}
			context.write(key, new Text(total));
		}
	}

	public static void main(String args[]) throws Exception
	{
		Path input = new Path(args[0]);
		Path output = new Path(args[1]);

		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf,"CountryCount");

		
		job.setJobName("CountCounters");
		job.setJar("CountCounters.jar");
		job.setJarByClass(ReadCSV.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		job.setMapperClass(CountMapper.class);
		job.setReducerClass(CountReducer.class);
		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class); 
		job.setNumReduceTasks(1);
		 
		FileInputFormat.setInputPaths(job, input);
		FileOutputFormat.setOutputPath(job, output);

		System.exit(job.waitForCompletion(true) ? 0 : 1);

		
		
	}

}