Changes between Version 8 and Version 9 of waue/2009/0822

Timestamp:

Aug 25, 2009, 2:02:37 PM (15 years ago)

Author:

waue

Comment:

--

waue/2009/0822

@@ -270 +270 @@
 
 import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.io.Text;
-import org.apache.hadoop.mapreduce.Mapper;
-
-public class mapper 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);
-                }
-        }
-}
+import org.apache.hadoop.mapred.MapReduceBase;
+import org.apache.hadoop.mapred.Mapper;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reporter;
+
+public class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable> {
+    private final static IntWritable one = new IntWritable(1);
+    private Text word = new Text();
+
+    public void map(LongWritable key, Text value, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
+      String line = value.toString();
+      StringTokenizer tokenizer = new StringTokenizer(line);
+      while (tokenizer.hasMoreTokens()) {
+        word.set(tokenizer.nextToken());
+        output.collect(word, one);
+      }
+    }
+  }
+
 }}}
 
@@ -321 +325 @@
 
 import java.io.IOException;
+import java.util.Iterator;
 
 import org.apache.hadoop.io.IntWritable;
 import org.apache.hadoop.io.Text;
-import org.apache.hadoop.mapreduce.Reducer;
-
-public class reducer 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);
-        }
-}
+import org.apache.hadoop.mapred.MapReduceBase;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.Reporter;
+
+public class Reduce extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable> {
+    public void reduce(Text key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
+      int sum = 0;
+      while (values.hasNext()) {
+        sum += values.next().get();
+      }
+      output.collect(key, new IntWritable(sum));
+    }
+  }
 }}}
 
@@ -368 +372 @@
 #!java
 package Sample;
-
-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.lib.input.FileInputFormat;
-import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
-import org.apache.hadoop.util.GenericOptionsParser;
+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.TextInputFormat;
+import org.apache.hadoop.mapred.TextOutputFormat;
 
 public class WordCount {
 
-        public static void main(String[] args) throws Exception {
-                Configuration conf = new Configuration();
-                String[] otherArgs = new GenericOptionsParser(conf, args)
-                                .getRemainingArgs();
-                if (otherArgs.length != 2) {
-                        System.err.println("Usage: wordcount <in> <out>");
-                        System.exit(2);
-                }
-                Job job = new Job(conf, "word count");
-                job.setJarByClass(WordCount.class);
-                job.setMapperClass(mapper.class);
-
-                job.setCombinerClass(reducer.class);
-                job.setReducerClass(reducer.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);
-        }
+   public static void main(String[] args) throws Exception {
+     JobConf conf = new JobConf(WordCount.class);
+     conf.setJobName("wordcount");
+
+     conf.setOutputKeyClass(Text.class);
+     conf.setOutputValueClass(IntWritable.class);
+
+     conf.setMapperClass(Map.class);
+     conf.setCombinerClass(Reduce.class);
+     conf.setReducerClass(Reduce.class);
+
+     conf.setInputFormat(TextInputFormat.class);
+     conf.setOutputFormat(TextOutputFormat.class);
+
+     FileInputFormat.setInputPaths(conf, new Path(args[0]));
+     FileOutputFormat.setOutputPath(conf, new Path(args[1]));
+
+     JobClient.runJob(conf);
+   }
 }
 }}}