Example9:   Emp9.java
___________________________________________________________________________________________________

[training@localhost ~]$ hadoop fs -cat mr/emp
101,ravi,10000,hyd,m,11
102,rani,12000,pune,f,12
103,ravina,13000,hyd,f,13
104,rana,14000,hyd,m,11
105,roopa,15000,pune,f,12
106,razeena,16000,pune,f,12
107,susma,14000,hyd,f,12
108,sampurnesh,20000,delhi,m,13
109,samantha,18000,pune,f,12
110,kamal,19000,delhi,m,11
111,krupa,21000,delhi,m,11
112,kapoor,16000,pune,m,12

schema : ecode, name, sal, city, sex, dno

Task:     number of employees in the   organization

Sql :   select   count(*)  from emp;
_________
Emp9.java
_________

package bharath.sreeram.big.halitcs;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.IOException;

public class Emp9
{
public static class MyMap extends Mapper<LongWritable,Text,Text,IntWritable>
{
public void map(LongWritable k,Text v, Context con)
throws IOException, InterruptedException
{
con.write(new Text("BigHalitics"), new IntWritable(1));
}
}
public static class MyRed extends Reducer<Text,IntWritable,IntWritable,Text>
{
public void reduce(Text k, Iterable<IntWritable> vlist, Context con)
throws IOException , InterruptedException
{
int tot=0;
for(IntWritable v:vlist)
{
  tot+=v.get();
}

con.write(new IntWritable(tot), new Text());
}
}
public static void main(String[] args) throws Exception
{
Configuration c = new Configuration();
Job j= new Job(c,"Test");
j.setJarByClass(Emp9.class);
j.setMapperClass(MyMap.class);
j.setReducerClass(MyRed.class);
j.setOutputKeyClass(Text.class);
j.setOutputValueClass(IntWritable.class);
Path p1 = new Path(args[0]);
Path p2 = new Path(args[1]);
   FileInputFormat.addInputPath(j,p1);
   FileOutputFormat.setOutputPath(j,p2);
   System.exit(j.waitForCompletion(true) ? 0:1);
}

}


Submitting Hadoop Job:

$ hadoop jar Desktop/mr.jar  bharath.sreeram.big.halitcs.Emp9 mr/emp  Result9

Output of the Job:

$ hadoop fs -cat Result9/part-r-00000
 12

Example 8:    Emp8.java
_____________________________________________________________________________________

[training@localhost ~]$ hadoop fs -cat mr/emp
101,ravi,10000,hyd,m,11
102,rani,12000,pune,f,12
103,ravina,13000,hyd,f,13
104,rana,14000,hyd,m,11
105,roopa,15000,pune,f,12
106,razeena,16000,pune,f,12
107,susma,14000,hyd,f,12
108,sampurnesh,20000,delhi,m,13
109,samantha,18000,pune,f,12
110,kamal,19000,delhi,m,11
111,krupa,21000,delhi,m,11
112,kapoor,16000,pune,m,12

schema : ecode, name, sal, city, sex, dno

Task:     Minimum  salary  of  the  organization

Sql :   select   min(sal) from emp;

_________
Emp8.java
_________

package bharath.sreeram.big.halitcs;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.IOException;

public class Emp8
{
public static class MyMap extends Mapper<LongWritable,Text,Text,IntWritable>
{
public void map(LongWritable k,Text v, Context con)
throws IOException, InterruptedException
{
String line = v.toString();
String[] w=line.split(",");
int sal=Integer.parseInt(w[2]);
con.write(new Text("BigHalitics"), new IntWritable(sal));
}
}
public static class MyRed extends Reducer<Text,IntWritable,IntWritable,Text>
{
public void reduce(Text k, Iterable<IntWritable> vlist, Context con)
throws IOException , InterruptedException
{
int min=0;
int cnt=0;
for(IntWritable v:vlist)
{
int sal=v.get();
cnt++;
if(cnt==1) min=sal;
  min=Math.min(min, sal);
}

con.write(new IntWritable(min), new Text());
}
}
public static void main(String[] args) throws Exception
{
Configuration c = new Configuration();
Job j= new Job(c,"Test");
j.setJarByClass(Emp8.class);
j.setMapperClass(MyMap.class);
j.setReducerClass(MyRed.class);
j.setOutputKeyClass(Text.class);
j.setOutputValueClass(IntWritable.class);
Path p1 = new Path(args[0]);
Path p2 = new Path(args[1]);
   FileInputFormat.addInputPath(j,p1);
   FileOutputFormat.setOutputPath(j,p2);
   System.exit(j.waitForCompletion(true) ? 0:1);
}

}

Submitting Job:


$ hadoop jar Desktop/mr.jar  bharath.sreeram.big.halitcs.Emp8 mr/emp  Result8

Output of the Job:

$ hadoop fs -cat Result8/part-r-00000
10000

Example 7:    Emp7.java
_______________________________________________________________________________________________


[training@localhost ~]$ hadoop fs -cat mr/emp
101,ravi,10000,hyd,m,11
102,rani,12000,pune,f,12
103,ravina,13000,hyd,f,13
104,rana,14000,hyd,m,11
105,roopa,15000,pune,f,12
106,razeena,16000,pune,f,12
107,susma,14000,hyd,f,12
108,sampurnesh,20000,delhi,m,13
109,samantha,18000,pune,f,12
110,kamal,19000,delhi,m,11
111,krupa,21000,delhi,m,11
112,kapoor,16000,pune,m,12

schema : ecode, name, sal, city, sex, dno

Task:     Maximum  salary  of  the  organization

Sql :   select    max(sal) from emp;

________
Emp7.java
________

package bharath.sreeram.big.halitcs;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.IOException;

public class Emp7
{
public static class MyMap extends Mapper<LongWritable,Text,Text,IntWritable>
{
public void map(LongWritable k,Text v, Context con)
throws IOException, InterruptedException
{
String line = v.toString();
String[] w=line.split(",");
int sal=Integer.parseInt(w[2]);
con.write(new Text("BigHalitics"), new IntWritable(sal));
}
}
public static class MyRed extends Reducer<Text,IntWritable,IntWritable,Text>
{
public void reduce(Text k, Iterable<IntWritable> vlist, Context con)
throws IOException , InterruptedException
{
int max=0;
for(IntWritable v:vlist)
{
  max=Math.max(max, v.get());
}

con.write(new IntWritable(max), new Text());
}
}
public static void main(String[] args) throws Exception
{
Configuration c = new Configuration();
Job j= new Job(c,"Test");
j.setJarByClass(Emp7.class);
j.setMapperClass(MyMap.class);
j.setReducerClass(MyRed.class);
j.setOutputKeyClass(Text.class);
j.setOutputValueClass(IntWritable.class);
Path p1 = new Path(args[0]);
Path p2 = new Path(args[1]);
   FileInputFormat.addInputPath(j,p1);
   FileOutputFormat.setOutputPath(j,p2);
   System.exit(j.waitForCompletion(true) ? 0:1);
}

}


Submitting Hadoop Job:

$ hadoop jar Desktop/mr.jar  bharath.sreeram.big.halitcs.Emp7 mr/emp  Result7

Output of the Job:

$ hadoop fs -cat Result7/part-r-00000
21000

Example 6  :   Emp6.java
_____________________________________________________________________________________________________

[training@localhost ~]$ hadoop fs -cat mr/emp
101,ravi,10000,hyd,m,11
102,rani,12000,pune,f,12
103,ravina,13000,hyd,f,13
104,rana,14000,hyd,m,11
105,roopa,15000,pune,f,12
106,razeena,16000,pune,f,12
107,susma,14000,hyd,f,12
108,sampurnesh,20000,delhi,m,13
109,samantha,18000,pune,f,12
110,kamal,19000,delhi,m,11
111,krupa,21000,delhi,m,11
112,kapoor,16000,pune,m,12

schema : ecode, name, sal, city, sex, dno

Task:     Average  salary of the organization

Sql :   select    avg(sal) from emp;

____________
Emp6.java
____________

package bharath.sreeram.big.halitcs;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.IOException;

public class Emp6
{
public static class MyMap extends Mapper<LongWritable,Text,Text,IntWritable>
{
public void map(LongWritable k,Text v, Context con)
throws IOException, InterruptedException
{
String line = v.toString();
String[] w=line.split(",");
int sal=Integer.parseInt(w[2]);
con.write(new Text("BigHalitics"), new IntWritable(sal));
}
}
public static class MyRed extends Reducer<Text,IntWritable,IntWritable,Text>
{
public void reduce(Text k, Iterable<IntWritable> vlist, Context con)
throws IOException , InterruptedException
{
int tot =0;
int cnt=0;
for(IntWritable v:vlist)
{
tot+=v.get();
   cnt++;
}
int avg=tot/cnt;
con.write(new IntWritable(avg), new Text());
}
}
public static void main(String[] args) throws Exception
{
Configuration c = new Configuration();
Job j= new Job(c,"Test");
j.setJarByClass(Emp6.class);
j.setMapperClass(MyMap.class);
j.setReducerClass(MyRed.class);
j.setOutputKeyClass(Text.class);
j.setOutputValueClass(IntWritable.class);
Path p1 = new Path(args[0]);
Path p2 = new Path(args[1]);
   FileInputFormat.addInputPath(j,p1);
   FileOutputFormat.setOutputPath(j,p2);
   System.exit(j.waitForCompletion(true) ? 0:1);
}

}

Submitting Hadoop Job:

$ hadoop jar Desktop/mr.jar  bharath.sreeram.big.halitcs.Emp6 mr/emp  Result6


Output Of the Job:

$ hadoop fs -cat Result6/part-r-00000
15666

Example : 5        Emp5.java
___________________________________________________________________________________________

[training@localhost ~]$ hadoop fs -cat mr/emp
101,ravi,10000,hyd,m,11
102,rani,12000,pune,f,12
103,ravina,13000,hyd,f,13
104,rana,14000,hyd,m,11
105,roopa,15000,pune,f,12
106,razeena,16000,pune,f,12
107,susma,14000,hyd,f,12
108,sampurnesh,20000,delhi,m,13
109,samantha,18000,pune,f,12
110,kamal,19000,delhi,m,11
111,krupa,21000,delhi,m,11
112,kapoor,16000,pune,m,12

schema : ecode, name, sal, city, sex, dno

Task:     Total salary of the organization

Sql :   select    sum(sal) from emp;
_________
Emp5.java
__________

package bharath.sreeram.big.halitcs;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.IOException;

public class Emp5
{
public static class MyMap extends Mapper<LongWritable,Text,Text,IntWritable>
{
public void map(LongWritable k,Text v, Context con)
throws IOException, InterruptedException
{
String line = v.toString();
String[] w=line.split(",");
int sal=Integer.parseInt(w[2]);
con.write(new Text("BigHalitics"), new IntWritable(sal));
}
}
public static class MyRed extends Reducer<Text,IntWritable,IntWritable,Text>
{
public void reduce(Text k, Iterable<IntWritable> vlist, Context con)
throws IOException , InterruptedException
{
int tot =0;
for(IntWritable v:vlist)
tot+=v.get();
con.write(new IntWritable(tot), new Text());
}
}
public static void main(String[] args) throws Exception
{
Configuration c = new Configuration();
Job j= new Job(c,"Test");
j.setJarByClass(Emp5.class);
j.setMapperClass(MyMap.class);
j.setReducerClass(MyRed.class);
j.setOutputKeyClass(Text.class);
j.setOutputValueClass(IntWritable.class);
Path p1 = new Path(args[0]);
Path p2 = new Path(args[1]);
   FileInputFormat.addInputPath(j,p1);
   FileOutputFormat.setOutputPath(j,p2);
   System.exit(j.waitForCompletion(true) ? 0:1);
}

}

Submitting Hadoop Job:

$ hadoop jar Desktop/mr.jar  bharath.sreeram.big.halitcs.Emp5 mr/emp  Result5


Output of the Job:

$ hadoop fs -cat Result5/part-r-00000
188000

Example4:    Emp4.java
_____________________________________________________________________________________________

[training@localhost ~]$ hadoop fs -cat mr/emp
101,ravi,10000,hyd,m,11
102,rani,12000,pune,f,12
103,ravina,13000,hyd,f,13
104,rana,14000,hyd,m,11
105,roopa,15000,pune,f,12
106,razeena,16000,pune,f,12
107,susma,14000,hyd,f,12
108,sampurnesh,20000,delhi,m,13
109,samantha,18000,pune,f,12
110,kamal,19000,delhi,m,11
111,krupa,21000,delhi,m,11
112,kapoor,16000,pune,m,12

schema : ecode, name, sal, city, sex, dno

Task:   for each  sex group ,  Number of employees

Sql :   select  sex  ,   count(*)  from emp group by  sex;
___________
Emp4.java
___________
package bharath.sreeram.big.halitcs;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.IOException;

public class Emp4
{
public static class MyMap extends Mapper<LongWritable,Text,Text,IntWritable>
{
public void map(LongWritable k,Text v, Context con)
throws IOException, InterruptedException
{
String line = v.toString();
String[] w=line.split(",");
String sex=w[4];
con.write(new Text(sex), new IntWritable(1));
}
}
public static class MyRed extends Reducer<Text,IntWritable,Text,IntWritable>
{
public void reduce(Text k, Iterable<IntWritable> vlist, Context con)
throws IOException , InterruptedException
{
int tot =0;
for(IntWritable v:vlist)
tot+=v.get();
con.write(k,new IntWritable(tot));
}
}
public static void main(String[] args) throws Exception
{
Configuration c = new Configuration();
Job j= new Job(c,"Test");
j.setJarByClass(Emp4.class);
j.setMapperClass(MyMap.class);
j.setReducerClass(MyRed.class);
j.setOutputKeyClass(Text.class);
j.setOutputValueClass(IntWritable.class);
Path p1 = new Path(args[0]);
Path p2 = new Path(args[1]);
   FileInputFormat.addInputPath(j,p1);
   FileOutputFormat.setOutputPath(j,p2);
   System.exit(j.waitForCompletion(true) ? 0:1);
}

}


Submitting Hadoop Job :

$ hadoop jar Desktop/mr.jar  bharath.sreeram.big.halitcs.Emp4 mr/emp  Result4


Output  of the Job :

$ hadoop fs -cat Result4/part-r-00000
f       6
m       6

Example3 :     Emp3.java
_________________________________________________________________________________

[training@localhost ~]$ hadoop fs -cat mr/emp
101,ravi,10000,hyd,m,11
102,rani,12000,pune,f,12
103,ravina,13000,hyd,f,13
104,rana,14000,hyd,m,11
105,roopa,15000,pune,f,12
106,razeena,16000,pune,f,12
107,susma,14000,hyd,f,12
108,sampurnesh,20000,delhi,m,13
109,samantha,18000,pune,f,12
110,kamal,19000,delhi,m,11
111,krupa,21000,delhi,m,11
112,kapoor,16000,pune,m,12

schema : ecode, name, sal, city, sex, dno

Task:   for each department  number , what is average salary.

Sql :   select  dno  , avg(sal) from emp group by  dno;


__________
Emp3.java
__________
package bharath.sreeram.big.halitcs;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.IOException;

public class Emp3
{
public static class MyMap extends Mapper<LongWritable,Text,Text,IntWritable>
{
public void map(LongWritable k,Text v, Context con)
throws IOException, InterruptedException
{
String line = v.toString();
String[] w=line.split(",");
String dno=w[5];
int sal=Integer.parseInt(w[2]);
con.write(new Text(dno), new IntWritable(sal));
}
}
public static class MyRed extends Reducer<Text,IntWritable,Text,IntWritable>
{
public void reduce(Text k, Iterable<IntWritable> vlist, Context con)
throws IOException , InterruptedException
{
int tot =0;
int cnt=0;
for(IntWritable v:vlist)
{
tot+=v.get();
cnt++;
}
int avg = tot/cnt;
con.write(k,new IntWritable(avg));
}
}
public static void main(String[] args) throws Exception
{
Configuration c = new Configuration();
Job j= new Job(c,"Test");
j.setJarByClass(Emp3.class);
j.setMapperClass(MyMap.class);
j.setReducerClass(MyRed.class);
j.setOutputKeyClass(Text.class);
j.setOutputValueClass(IntWritable.class);
Path p1 = new Path(args[0]);
Path p2 = new Path(args[1]);
   FileInputFormat.addInputPath(j,p1);
   FileOutputFormat.setOutputPath(j,p2);
   System.exit(j.waitForCompletion(true) ? 0:1);
}

}


Submitting Hadoop Job:

$ hadoop jar Desktop/mr.jar  bharath.sreeram.big.halitcs.Emp3 mr/emp  Result3

output of the job:

$ hadoop fs -cat Result3/part-r-00000
11      16000
12      15166
13      16500

Example 2:      Emp2.java
_____________________________________________________________________________________

input Hdfs file:  mr/emp
____________________
[training@localhost ~]$ hadoop fs -cat mr/emp
101,ravi,10000,hyd,m,11
102,rani,12000,pune,f,12
103,ravina,13000,hyd,f,13
104,rana,14000,hyd,m,11
105,roopa,15000,pune,f,12
106,razeena,16000,pune,f,12
107,susma,14000,hyd,f,12
108,sampurnesh,20000,delhi,m,13
109,samantha,18000,pune,f,12
110,kamal,19000,delhi,m,11
111,krupa,21000,delhi,m,11
112,kapoor,16000,pune,m,12

schema : ecode, name, sal, city, sex, dno

Task:   for each city , what is total salary.

Sql :   select city , sum(sal) from emp group by  city;

__________
Emp1.java
__________



package bharath.sreeram.big.halitcs;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.IOException;

public class Emp2
{
public static class MyMap extends Mapper<LongWritable,Text,Text,IntWritable>
{
public void map(LongWritable k,Text v, Context con)
throws IOException, InterruptedException
{
String line = v.toString();
String[] w=line.split(",");
String city=w[3];
int sal=Integer.parseInt(w[2]);
con.write(new Text(city), new IntWritable(sal));
}
}
public static class MyRed extends Reducer<Text,IntWritable,Text,IntWritable>
{
public void reduce(Text k, Iterable<IntWritable> vlist, Context con)
throws IOException , InterruptedException
{
int tot =0;
for(IntWritable v:vlist)
tot+=v.get();
con.write(k,new IntWritable(tot));
}
}
public static void main(String[] args) throws Exception
{
Configuration c = new Configuration();
Job j= new Job(c,"Test");
j.setJarByClass(Emp2.class);
j.setMapperClass(MyMap.class);
j.setReducerClass(MyRed.class);
j.setOutputKeyClass(Text.class);
j.setOutputValueClass(IntWritable.class);
Path p1 = new Path(args[0]);
Path p2 = new Path(args[1]);
   FileInputFormat.addInputPath(j,p1);
   FileOutputFormat.setOutputPath(j,p2);
   System.exit(j.waitForCompletion(true) ? 0:1);
}

}


Submitting Hadoop Job:

$ hadoop jar Desktop/mr.jar  bharath.sreeram.big.halitcs.Emp2 mr/emp  Result2


output of the Job:

$ hadoop fs -cat Result2/part-r-00000
delhi   60000
hyd     51000
pune    77000

Example1.  Emp1.java
__________________________________________________________________________________________
input Hdfs file:  mr/emp
____________________
[training@localhost ~]$ hadoop fs -cat mr/emp
101,ravi,10000,hyd,m,11
102,rani,12000,pune,f,12
103,ravina,13000,hyd,f,13
104,rana,14000,hyd,m,11
105,roopa,15000,pune,f,12
106,razeena,16000,pune,f,12
107,susma,14000,hyd,f,12
108,sampurnesh,20000,delhi,m,13
109,samantha,18000,pune,f,12
110,kamal,19000,delhi,m,11
111,krupa,21000,delhi,m,11
112,kapoor,16000,pune,m,12

schema : ecode, name, sal, city, sex, dno

Task:   for each sex group , what is total salary.

Sql :   select sex, sum(sal) from emp group by sex;

__________
Emp1.java
__________

package bharath.sreeram.big.halitcs;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.IOException;

public class Emp1
{
public static class MyMap extends Mapper<LongWritable,Text,Text,IntWritable>
{
public void map(LongWritable k,Text v, Context con)
throws IOException, InterruptedException
{
String line = v.toString();
String[] w=line.split(",");
String sex=w[4];
int sal=Integer.parseInt(w[2]);
con.write(new Text(sex), new IntWritable(sal));
}
}
public static class MyRed extends Reducer<Text,IntWritable,Text,IntWritable>
{
public void reduce(Text k, Iterable<IntWritable> vlist, Context con)
throws IOException , InterruptedException
{
int tot =0;
for(IntWritable v:vlist)
tot+=v.get();
con.write(k,new IntWritable(tot));
}
}
public static void main(String[] args) throws Exception
{
Configuration c = new Configuration();
Job j= new Job(c,"Test");
j.setJarByClass(Emp1.class);
j.setMapperClass(MyMap.class);
j.setReducerClass(MyRed.class);
j.setOutputKeyClass(Text.class);
j.setOutputValueClass(IntWritable.class);
Path p1 = new Path(args[0]);
Path p2 = new Path(args[1]);
   FileInputFormat.addInputPath(j,p1);
   FileOutputFormat.setOutputPath(j,p2);
   System.exit(j.waitForCompletion(true) ? 0:1);
}

}

Submitting Hadoop job:

$ hadoop jar Desktop/mr.jar  bharath.sreeram.big.halitcs.Emp1 mr/emp  Result1


output of the job:

$ hadoop fs -cat Result1/part-r-00000
f       88000
m       100000