MapReduce之WordCount案例

MapReduce之WordCount案例 前言

学习大数据框架通常都是从wordcount案例开始的，也是学习框架的基础，wordcount虽然简单，如果能彻底搞清楚其运行原理，对后续深入学习和掌握MapReduce非常有帮助的，本篇以一个wordcount为例，通过代码演示下wordcount的编码过程

环境准备

1. windows下hadoop的配置环境，需要在windows环境下配置一下环境变量
2. linux下的hdfs运行环境
3. 一个单词统计的文本文件，文件内容格式如下

关羽 关羽
赵云 
刘备 刘备
黄盖
张飞
马超
魏延

通过WordCount程序，我们期望最终的输出结果格式为：

关羽 2
赵云 1
刘备 2
…

编码步骤

MapReduce的编码风格比较套路化，一般来说，最简单的WordCount只需要3个类即可，一个继承Mapper类的自定义Map类，一个继承Reducer类的自定义Reduce类，以及一个将这两个类聚合在一起并执行job的driver类即可

很多学习MapReduce的同学比较头疼和容易犯迷糊的地方是自定义的Map 和 Reduce类里面的参数，以及Map 和 Reduce 类中的重写方法的代码执行逻辑，基本上来说，把这两个问题搞明白了，遇到大部分的业务场景，都可以迎刃而解了

下面看具体的编码过程

1、导入依赖

		
            org.apache.hadoop
            hadoop-client
            3.1.3
        

        
            junit
            junit
            4.12
        

        
            org.slf4j
            slf4j-log4j12
            1.7.30
        

2、resources目录下添加一个log4j.properties文件

log4j.rootLogger=INFO, stdout  
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/spring.log  
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout  
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

3、自定义 Map类

需要提供一个继承Mapper类的自定义Map类，并重写里面的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 DemoMapper extends Mapper {

    private Text outKey = new Text();

    private IntWritable outVal = new IntWritable();

    
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 张飞 张飞
        String linevalue = value.toString();
        String[] lineWords = linevalue.split(" ");
        //将数据循环写出
        for (String lineWord : lineWords) {
            outKey.set(lineWord);
            outVal.set(1);
            context.write(outKey,outVal);
        }
    }

}

3、自定义 Reduce类

需要提供一个继承Reducer类的自定义Reducer类，并重写里面的reduce方法，需要重点关注几个核心参数的解释说明

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;


public class DemoReducer extends Reducer {

    private IntWritable outVal = new IntWritable();

    
    @Override
    protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
        int sum = 0;
        for(IntWritable value : values){
            sum +=value.get();
        }
        outVal.set(sum);
        context.write(key,outVal);
    }

}

4、job类

通过job类，将上面的自定义map和reduce类进行关联，并配置一些基本的输出参数，基本上按照一个固定的模板格式编写即可

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;

public class DemoJobDriver {

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

        //1、获取job
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);

        //2、设置jar路径
        job.setJarByClass(DemoJobDriver.class);

        //3、关联mapper 和 Reducer
        job.setMapperClass(DemoMapper.class);
        job.setReducerClass(DemoReducer.class);

        //4、设置 map输出的 key/val 的类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        //5、设置最终输出的key / val 类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //6、设置最终的输出路径
        String inputPath = "F:\测试文件\hello.txt";
        String outPath = "F:\测试文件\result\hello_result.txt";
        FileInputFormat.setInputPaths(job,new Path(inputPath));
        FileOutputFormat.setOutputPath(job,new Path(outPath));

        // 7 提交job
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }

}

下面运行这段程序，观察下最后的输出结果，程序执行完毕后，在本地的输出目录下产生了几个文件，目标文件即为最后那个，可以打开看一下

以上即为我们想要的结果

5、提交到hadoop上运行

以上为本地运行job的结果，生产环境中，需要将开发的job打包，上传到hadoop进群去运行

修改下main程序中的job配置，这里的参数将由外部传入

关于WordCount案例，初学者在自定义的Map方法中的map逻辑里面，对如何读取外部输入的文件是如何流转的存在困惑

简单来说，就是什么时候进入到这个map方法中呢？通过继承的Mapper类，我们进入到源码中，在Mapper中，有一个很重要的run方法

我们不妨通过断点调试下这段程序，然后发现，这个while循环其实循环的就是一行行文本数据，每读取完毕一行数据，进入到 this.map方法，然后由 this.map 进入到自定义的Map类的map方法中执行自己的逻辑，依次进行，直到读取完毕所有的文本行数据，而自定义的Reduce逻辑亦是如此

本篇到此结束，最后感谢观看！