wordcount.java

 package com.biencloud.test;  
 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.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.output.FileOutputFormat;    
 import org.apache.hadoop.util.GenericOptionsParser;    
 
 public class WordCount {    
 
 
     public static class TokenizerMapper extends    
             Mapper<Object, Text, Text, IntWritable> {    
 
         private final static IntWritable one = new IntWritable(1);    
         private Text word = new Text();    
 
         public void map(Object key, Text value, Context context)    
                 throws IOException, InterruptedException {    
             StringTokenizer itr = new StringTokenizer(value.toString());    
             while (itr.hasMoreTokens()) {    
                word.set(itr.nextToken());    
                 context.write(word, one);    
             }    
         }    
     }    
 
     public static class IntSumReducer extends    
             Reducer<Text, IntWritable, Text, IntWritable> {    
        private IntWritable result = new IntWritable();    
 
         public void reduce(Text key, Iterable<IntWritable> values,    
                 Context context) throws IOException, InterruptedException {    
             int sum = 0;    
             for (IntWritable val : values) {    
                 sum += val.get();    
             }    
             result.set(sum);    
             context.write(key, result);    
         }    
     }    
 
     public static void main(String[] args) throws Exception {  
         Configuration conf = new Configuration();   
         conf.set("mapred.job.tracker", "192.168.0.118:9001");  
 //        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();    
 //        if (otherArgs.length != 2) {    
 //            System.err.println("Usage: wordcount <in> <out>");    
 //           System.exit(2);    
 //       }    
         String[] otherArgs = new String[2];  
         otherArgs[0] = "hdfs://192.168.0.118:9000/user/hadoop/f";  
         otherArgs[1] = "hdfs://192.168.0.118:9000/user/hadoop/f2";  
         conf.set("hadoop.job.ugi", "hadoop,hadoop");  
         Job job = new Job(conf, "word count");    
         job.setNumReduceTasks(4);  
         job.setJarByClass(WordCount.class);    
         job.setMapperClass(TokenizerMapper.class);    
         job.setCombinerClass(IntSumReducer.class);    
         job.setReducerClass(IntSumReducer.class);    
         job.setOutputKeyClass(Text.class);    
         job.setOutputValueClass(IntWritable.class);    
         FileInputFormat.addInputPath(job, new Path(otherArgs[0]));    
         FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));    
         System.exit(job.waitForCompletion(true) ? 0 : 1);    
 
 //        Configuration conf=new Configuration();  
 //        conf.set("hadoop.job.ugi", "hadoop,hadoop");  
 //        FileSystem src=FileSystem.getLocal(conf);  
 //        FileSystem dst= FileSystem.get(URI.create(hadoopFile),conf);  
 //        Path srcpath = new Path(localFile);  
 //        Path dstpath = new Path(hadoopFile);  
 //        FileUtil.copy(src, srcpath, dst, dstpath,false,conf);  
     }    
 }