Added Sample using old API and One Sample for custom formater

src/main/java/com/dsinpractice/samples/hadoop/mapred/charcount/CharCount.java

@@ -0,0 +1,62 @@
+package com.dsinpractice.samples.hadoop.mapred.charcount;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.FileInputFormat;
+import org.apache.hadoop.mapred.FileOutputFormat;
+import org.apache.hadoop.mapred.JobClient;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.MapReduceBase;
+import org.apache.hadoop.mapred.Mapper;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapred.TextInputFormat;
+import org.apache.hadoop.mapred.TextOutputFormat;
+
+public class CharCount {
+
+
+
+
+	//Write client which will run the job
+	public static void main(String[] args) {
+
+		//1. create job conf
+		JobConf conf = new JobConf(CharCount.class);
+		conf.setJobName("CharCount");
+
+		//2. set the output key and value type
+		conf.setMapOutputKeyClass(Text.class);
+		conf.setMapOutputValueClass(IntWritable.class);
+
+		//3. set the map and reduce class
+		conf.setMapperClass(CharCountMap.class);
+		conf.setReducerClass(CharCountReduce.class);
+
+		//4. set the inoput output data format
+		conf.setInputFormat(TextInputFormat.class);
+		conf.setOutputFormat(TextOutputFormat.class);
+
+		//5. set the input / output directories
+		//FileInputFormat.setInputPaths(conf, new Path("/hdfspath/data/wcinput/"));
+		//FileOutputFormat.setOutputPath(conf, new Path("/hdfspath/data/ccoutput"));
+
+        //5. set the input and output file path
+        FileInputFormat.setInputPaths(conf, new Path(args[0]));
+        FileOutputFormat.setOutputPath(conf, new Path(args[1]));
+
+        //6. run the job
+		try {
+			JobClient.runJob(conf);
+		} catch (IOException e) {
+			e.printStackTrace();
+		}
+	}
+
+}

src/main/java/com/dsinpractice/samples/hadoop/mapred/charcount/CharCountMap.java

@@ -0,0 +1,37 @@
+package com.dsinpractice.samples.hadoop.mapred.charcount;
+
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.MapReduceBase;
+import org.apache.hadoop.mapred.Mapper;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reporter;
+
+import java.io.IOException;
+
+/**
+ * Created by admin on 10/23/14.
+ */
+// write Map class
+public  class CharCountMap extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable> {
+
+
+    @Override
+    public void map(LongWritable key, Text value, OutputCollector<Text, IntWritable> output, Reporter reporter)
+            throws IOException {
+        // Logic to read line and count chars
+
+        String line = value.toString();
+
+        if (line == null || line.equals(""))
+            return;
+
+        for (int i = 0; i < line.length(); i++) {
+            value.set("" + line.charAt(i));
+            output.collect(value, new IntWritable(1));
+        }
+
+    }
+
+}

src/main/java/com/dsinpractice/samples/hadoop/mapred/charcount/CharCountReduce.java

@@ -0,0 +1,31 @@
+package com.dsinpractice.samples.hadoop.mapred.charcount;
+
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.*;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+/**
+ * Created by admin on 10/23/14.
+ */
+// write Map class
+//Write Reduce class
+public   class CharCountReduce extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable> {
+
+    @Override
+    public void reduce(Text key, Iterator<IntWritable> value, OutputCollector<Text, IntWritable> output, Reporter reporter)
+            throws IOException {
+        // Logic to collect and segregate the char counts per char
+
+        int sum = 0;
+
+        while(value.hasNext()){
+            sum = sum + value.next().get();
+        }
+        output.collect(key, new IntWritable(sum));
+    }
+
+}

src/main/java/com/dsinpractice/samples/hadoop/mapred/customformat/CustomFormatClient.java

@@ -0,0 +1,46 @@
+package com.dsinpractice.samples.hadoop.mapred.customformat;
+
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
+
+
+public class CustomFormatClient {
+
+
+
+	public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
+
+		/*if (args-Djava.security.krb5.realm=OX.AC.UK -Djava.security.krb5.kdc=kdc0.ox.ac.uk:kdc1.ox.ac.uk.length != 2) {
+		      System.err.println("Usage: <input path> <output path>");
+		      System.exit(-1);
+		}*/
+
+		Configuration conf = new Configuration();
+
+
+		Job job = new Job(conf,"Custom Input Type");
+		job.setJarByClass(CustomFormatClient.class);
+		job.setNumReduceTasks(0);
+
+		job.setMapperClass(MyMapper.class);
+		job.setOutputKeyClass(MyKey.class);
+		job.setOutputValueClass(MyValue.class);
+		job.setInputFormatClass(MyInputFormat.class);
+
+	    FileInputFormat.addInputPath(job, new Path(args[0]));
+	    FileOutputFormat.setOutputPath(job, new Path(args[1]));
+
+
+
+	    job.waitForCompletion(true);
+	}
+
+
+}

src/main/java/com/dsinpractice/samples/hadoop/mapred/customformat/MyInputFormat.java

@@ -0,0 +1,19 @@
+package com.dsinpractice.samples.hadoop.mapred.customformat;
+
+import java.io.IOException;
+
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.RecordReader;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
+
+
+public class MyInputFormat extends FileInputFormat<MyKey, MyValue>{
+
+	@Override
+	public RecordReader<MyKey, MyValue> createRecordReader(InputSplit split,
+			TaskAttemptContext context) throws IOException, InterruptedException {
+		return new MyRecordReader();
+	}
+
+}

src/main/java/com/dsinpractice/samples/hadoop/mapred/customformat/MyKey.java

@@ -0,0 +1,62 @@
+package com.dsinpractice.samples.hadoop.mapred.customformat;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.WritableComparable;
+
+public class MyKey implements WritableComparable {
+
+	private Text imei, deviceModel;
+
+	public MyKey() {
+		setImei(new Text());
+		setDeviceModel(new Text());
+	}
+
+	public MyKey(Text _imei, Text _deviceModel) {
+		setImei(_imei);
+		setDeviceModel(_deviceModel);
+	}
+
+	@Override
+	public void readFields(DataInput in) throws IOException {
+		getImei().readFields(in);
+		getDeviceModel().readFields(in);
+	}
+
+	@Override
+	public void write(DataOutput out) throws IOException {
+		getImei().write(out);
+		getDeviceModel().write(out);
+	}
+
+	@Override
+	public int compareTo(Object obj) {
+		MyKey argKey = (MyKey) obj;
+		int result = this.getImei().compareTo(argKey.getImei());
+		if(result != 0){
+			return result;
+		}
+		return this.getDeviceModel().compareTo(argKey.getDeviceModel());
+	}
+
+	public Text getImei() {
+		return imei;
+	}
+
+	public void setImei(Text imei) {
+		this.imei = imei;
+	}
+
+	public Text getDeviceModel() {
+		return deviceModel;
+	}
+
+	public void setDeviceModel(Text deviceModel) {
+		this.deviceModel = deviceModel;
+	}
+
+}

src/main/java/com/dsinpractice/samples/hadoop/mapred/customformat/MyMapper.java

@@ -0,0 +1,26 @@
+package com.dsinpractice.samples.hadoop.mapred.customformat;
+
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.Mapper;
+
+
+
+/**
+ * Created by admin on 10/23/14.
+ */
+public  class MyMapper extends Mapper<MyKey, MyValue, Text, Text> {
+
+
+    public void map(MyKey key, MyValue value, Context context) throws java.io.IOException ,InterruptedException {
+        String imei = key.getImei().toString();
+        String model = key.getDeviceModel().toString();
+        if(imei.startsWith("9765")){
+            String status = value.getStatus().toString();
+            String country = value.getCountry().toString();
+            String basePhone = value.getBasePhone().toString();
+            String advPhone = value.getAdvPhone().toString();
+            context.write(new Text(imei),new Text(country));
+        }
+
+    }
+}

src/main/java/com/dsinpractice/samples/hadoop/mapred/customformat/MyRecordReader.java

@@ -0,0 +1,74 @@
+package com.dsinpractice.samples.hadoop.mapred.customformat;
+
+import java.io.IOException;
+
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.RecordReader;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.mapreduce.lib.input.LineRecordReader;
+
+public class MyRecordReader extends RecordReader<MyKey, MyValue> {
+
+	private MyKey myKey;
+	private MyValue myValue;
+	private LineRecordReader reader = new LineRecordReader();
+
+	@Override
+	public void close() throws IOException {
+		// TODO Auto-generated method stub
+		reader.close();
+	}
+
+	@Override
+	public MyKey getCurrentKey() throws IOException, InterruptedException {
+		// TODO Auto-generated method stub
+		return myKey;
+	}
+
+	@Override
+	public MyValue getCurrentValue() throws IOException, InterruptedException {
+		// TODO Auto-generated method stub
+		return myValue;
+	}
+
+	@Override
+	public float getProgress() throws IOException, InterruptedException {
+		// TODO Auto-generated method stub
+		return reader.getProgress();
+	}
+
+	@Override
+	public void initialize(InputSplit split, TaskAttemptContext context)
+			throws IOException, InterruptedException {
+		// TODO Auto-generated method stub
+		reader.initialize(split, context);
+	}
+
+	@Override
+	public boolean nextKeyValue() throws IOException, InterruptedException {
+		boolean isNextValue = reader.nextKeyValue();
+		if (isNextValue) {
+			if (myKey == null) {
+				myKey = new MyKey();
+			}
+			if (myValue == null) {
+				myValue = new MyValue();
+			}
+			Text line = reader.getCurrentValue();
+			String[] tokens = line.toString().split(",");
+			myKey.setImei(new Text(tokens[0]));
+			myKey.setDeviceModel(new Text(tokens[1]));
+			myValue.setStatus(new Text(tokens[2]));
+			myValue.setCountry(new Text(tokens[3]));
+			myValue.setBasePhone(new Text(tokens[4]));
+			myValue.setAdvPhone(new Text(tokens[5]));
+
+		} else {
+			myKey = null;
+			myValue = null;
+		}
+		return isNextValue;
+	}
+
+}