public class MrMap  extends Mapper{

protected void map(Object key, Text value, Context context) {

String line= value.toString();

String[] words = line.split(" ");

for (String str : words) {

Text text=new Text(str);

IntWritable num=new IntWritable(1);

    try {

context.write(text, num);

} catch (Exception e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

}

   };

}

public class WordReduce extends Reducer {

protected void reduce(Text text, Iterable itrs, Context context) {

int sum = 0;

for (IntWritable itr : itrs) {

sum = sum + itr.get();

}

try {

context.write(text, new IntWritable(sum));

} catch (IOException e) {

// TODO Auto-generated catch block

e.printStackTrace();

} catch (InterruptedException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

};

}

public class WordCount {

/**

* @param args

* @throws IOException

* @throws InterruptedException

* @throws ClassNotFoundException

*/

public static void main(String[] args) throws IOException {

Configuration conf = new Configuration();

FileSystem fs = FileSystem.get(conf);

Job job = null;

try {

job = Job.getInstance(conf);

job.setJobName("wc");

job.setJarByClass(WordCount.class);

job.setMapperClass(WordMap.class);

job.setReducerClass(WordReduce.class);

job.setMapOutputKeyClass(Text.class);

job.setMapOutputValueClass(IntWritable.class);

FileInputFormat.addInputPath(job, new Path("/word.txt"));

if (fs.exists(new Path("/out"))) {

fs.delete(new Path("/out"));

}

FileOutputFormat.setOutputPath(job, new Path("/out"));

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

} catch  (Exception e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

}

}