Reducer阶段

log4j.rootLogger=INFO, stdout, logfile
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.File=target/wordcount.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

完成后在包下创建三个java类

WordCountMapper类

在WordCountMapper类中继承MapReduce类

参数解读：

KEYIN ：map阶段输入的key类型：LongWritable

VALUEIN：map阶段输入的value类型：Text

KEYOUT：map阶段输出的key类型：Text

VALUEOUT：map阶段输入的value类型：IntWritable

 重写一个map方法

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class WordCountMapper extends Mapper<LongWritable, Text,Text, IntWritable> {//继承MapReduce类(泛型)
    private Text outKey = new Text(); //定义输出的key值
    private IntWritable outVlue = new IntWritable(1); //不进行聚合操作默认为1

    //重写一个map方法
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        //获取一行
        String line = value.toString();

        //切割
        String[] words = line.split(" "); //每一行单词都放到键中 遇到空格则分割

        //循环写出
        for (String word : words) {
            //封装outKey
            outKey.set(word);

            //写出
            context.write(outKey,outVlue); //参数为输出的KEY和VALUE
        }
    }
}

WordCountReducer类

在WordCountReducer类中继承Reducer类

参数解读：

KEYIN ：reduce阶段输入的key类型：Text

VALUEIN：reduce阶段输入的value类型：IntWritable

KEYOUT：reduce阶段输出的key类型：Text

VALUEOUT：reduce阶段输入的value类型：IntWritable

 重写reduce方法

import org.apache.hadoop.io.IntWritable;
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, IntWritable,Text, IntWritable> { //继承Reducer类(泛型)
    //封装
    private IntWritable outValue = new IntWritable();

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {

        //定义变量累加
        int sum = 0;

        //要统计的单词(1,1)
        //累加
        for (IntWritable value : values) {

            sum += value.get(); //获取

        }

        //变换类型

        outValue.set(sum);

        //写出
        context.write(key,outValue); //outValue就是转换后的sum
    }
}

WordCountDriverr 驱动类

基本流程：

1. 获取job 设置jar包路径
2. 关联Mapper、Reducer
3. 设置map输出的k,v类型
4. 最终输出的k,v类型
5. 设置输入路径和输出路径
6. 提交job