MapReduce Flights between Origin and Destination

MapReduce Flights between Origin and Destination:

package com.lbrce.flight;
import java.io.IOException;
import java.util.Iterator;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.conf.Configuration;
public class Flight {
public static class FlightMapper extends Mapper<LongWritable, Text, Text, Text>
{
public void map(LongWritable arg0, Text Value, Context context) throws IOException, InterruptedException
{
String line = Value.toString();
if (!(line.length() == 0))
{
String fno = line.substring(0, 4);
String origin=line.substring(8, 12).trim();
String dest =line.substring(13, 18).trim();
if(origin.equals("HYD")&&dest.equals("SAN"))
{
context.write(new Text("Flight " + fno),new Text("HYD SAN"));
}
}
}
}
public static class FlightReducer extends Reducer<Text, Text, Text, Text>
{
public void reduce(Text Key, Iterator<Text> Values, Context context)throws IOException, InterruptedException
{
String nof = Values.next().toString();
context.write(Key, new Text(nof));
}

}
public static void main(String[] args) throws Exception
{
Configuration conf = new Configuration();
Job job = new Job(conf, "Flight example");
job.setJarByClass(Flight.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);
job.setMapperClass(FlightMapper.class);
job.setReducerClass(FlightReducer.class);
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
Path OutputPath = new Path(args[1]);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}

Sample Input:

Flight Num        Origin     Destination    Arrival Time

----------------------------------------------------------------

AI111                 HYD         SAN             22:30

QA222               BOM         NEY             24:26

SA333                DEL          DAL             32:24

BA444               CHE         SAN             42:15

SA555                HYD         NEJ             24:26

QA666               BAN          DAL            22:30

AI777                 HYD         SAN             32:24

SA888                 DEL         SAN             42:15

BA999                 BAN         NEY             32:24

SA123                 BOM         NEJ             24:26

QA321                 CHE         SAN             42:15

SA345                 BAN         DAL             24:26

AI456                 CHE         SAN               42:15

BA789                 HYD         SAN             22:30

QA156                 BOM         NEJ             32:24

SA234                 BAN         DAL             24:26

BA132                 BOM         NEJ             42:15

AI431                 HYD         SAN             22:30

AA001                 CHE         SAN             32:24

AA007                 BOM         NEJ             24:26

AA009                 HYD         SAN             24:26

DT876                 BAN         DAL             42:15

JT567                 HYD          SAN             22:30