作业五【MapReduce词频统计】

未分类

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>njt.song.study</groupId>
  <artifactId>hadoop</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <packaging>jar</packaging>
  <name>hadoop</name>
  <url>http://maven.apache.org</url>
  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  </properties>
  <dependencies>
        <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-hdfs</artifactId>
        <version>2.5.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-common</artifactId>
        <version>2.5.1</version>
    </dependency>   
    <dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-mapreduce-client-core</artifactId>
    <version>2.5.1</version>
</dependency>
    <dependency>
      <groupId>junit</groupId>
      <artifactId>junit</artifactId>
      <version>3.8.1</version>
      <scope>test</scope>
    </dependency>
  </dependencies>
</project>

package njt.song.study.hadoop;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;  
import org.apache.hadoop.conf.Configured;
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.Tool;
import org.apache.hadoop.util.ToolRunner;
public class WordCount extends Configured implements Tool {
      public static class WordCountMapper extends Mapper<Object,Text,Text,IntWritable> {  
            private static final IntWritable inputKey =  new IntWritable(1);  
            private Text outPutValue = new Text();  
            protected void map(Object key, Text value, Context context)  
                    throws IOException, InterruptedException {  
                String line = value.toString();  
                String[] words = line.split("\t");
                for(String word:words){
                    outPutValue.set(word);
                    context.write(outPutValue, inputKey);
                }
            }  
        }  
      public static class WordCountReducer extends Reducer<Text,IntWritable,Text,IntWritable> {  
            private IntWritable total = new IntWritable();  
            @Override  
            protected void reduce(Text key, Iterable<IntWritable> values,Context context)  
                    throws IOException, InterruptedException {  
               int sum = 0;  
               for(IntWritable value:values){
                   sum += value.get();         
               }
               total.set(sum);  
               context.write(key,total);   
            }  
        }  
      public static void main(String[] args) throws Exception{  
        Configuration conf = new Configuration(); 
        int status = ToolRunner.run(conf, new WordCount(),args);
        System.exit(status);  
      }
      public int run(String[] args) throws Exception {
          Configuration configuration = super.getConf();
          Job job = Job.getInstance(configuration,this.getClass().getSimpleName());
          job.setJarByClass(WordCount.class);
          Path inPath = new Path(args[0]);
          FileInputFormat.addInputPath(job, inPath);
          job.setMapperClass(WordCountMapper.class);
          job.setMapOutputKeyClass(Text.class);
          job.setMapOutputValueClass(IntWritable.class);
          job.setReducerClass(WordCountReducer.class);
          job.setOutputKeyClass(Text.class);
          job.setOutputValueClass(IntWritable.class);
          Path outPath = new Path(args[1]);
          FileOutputFormat.setOutputPath(job, outPath);
          boolean isSuccess = job.waitForCompletion(true);
          return isSuccess ? 0 : 1;
        }         
}

bin/yarn jar /home/sjf/myWordcount.jar /input/word.txt /output

查看结果

bin/hdfs dfs -ls /output

bin/hdfs dfs cat /output/part-r-00000