Java编写Mapreduce程序过程浅析

< dependencies >

         < dependency >

             < groupId >org.apache.hadoop</ groupId >

             < artifactId >hadoop-client</ artifactId >

             < version >3.3.0</ version >

         </ dependency >

         < dependency >

             < groupId >junit</ groupId >

             < artifactId >junit</ artifactId >

             < version >4.12</ version >

             < scope >test</ scope >

         </ dependency >

         < dependency >

             < groupId >org.slf4j</ groupId >

             < artifactId >slf4j-api</ artifactId >

             < version >1.7.30</ version >

         </ dependency >

         < dependency >

             < groupId >junit</ groupId >

             < artifactId >junit</ artifactId >

             < version >3.8.2</ version >

         </ dependency >

         < dependency >

             < groupId >junit</ groupId >

             < artifactId >junit</ artifactId >

             < version >4.12</ version >

             < scope >compile</ scope >

         </ dependency >

     </ dependencies >

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

import org.apache.hadoop.util.StringUtils;

import java.io.IOException;

public class WordCountMapper extends Mapper<LongWritable,Text,Text,LongWritable> {

     @Override

     protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

         //如果当前数据不为空

         if (value!= null ){

             //获取每一行的数据

             String line = value.toString();

             //将一行数据根据空格分开

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

             String[] words = StringUtils.split(line, ' ' ); //hadoop的StringUtils.split方法对大数据来说比Java自带的拥有更好的性能

             //输出键值对

             for (String word : words) {

                 context.write( new Text(word), new LongWritable( 1 ));

             }

         }

     }

}

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class WordCountReducer extends Reducer<Text, LongWritable,Text,LongWritable> {

     @Override

     protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {

         //累加单词的数量

         long sum = 0 ;

         //遍历单词计数数组,将值累加到sum中

         for (LongWritable value : values) {

             sum += value.get();

         }

         //输出每次最终的计数结果

         context.write(key, new LongWritable(sum));

     }

}

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.conf.Configured;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.LongWritable;

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.Tool;

import org.apache.hadoop.util.ToolRunner;

public class WordCountRunner extends Configured implements Tool {

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

         ToolRunner.run( new Configuration(), new WordCountRunner(),args);

     }

     @Override

     public int run(String[] args) throws Exception {

         Configuration conf = new Configuration();

         Job job = Job.getInstance(conf, "word count" );

         job.setJarByClass(WordCountRunner. class );

         job.setMapperClass(WordCountMapper. class );

         job.setReducerClass(WordCountReducer. class );

         job.setMapOutputKeyClass(Text. class );

         job.setMapOutputValueClass(LongWritable. class );

         job.setOutputKeyClass(Text. class );

         job.setOutputValueClass(LongWritable. class );

         //设置统计文件输入的路径,将命令行的第一个参数作为输入文件的路径

         //读取maven项目下resources目录的文档   

         String path = getClass().getResource( "/words.txt" ).getPath();

         FileInputFormat.setInputPaths(job,path);

         //设置结果数据存放路径,将命令行的第二个参数作为数据的输出路径

         //输出目录必须不存在!!!

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

         return job.waitForCompletion( true ) ? 0 : 1 ;

     }

}