WordCount代码详解

本文简要讲解MapReduce实例–WordCount代码

使用MapReduce的前提：

• 待处理的数据集可以分解成许多小的数据集
• 每一个小数据集都可以完全并行地进行处理

WordCount是最简单、最能体现MapReduce思想的程序之一，可以称为MapReduce版Hello World，其主要功能是：统计一系列文本文件中每个单词出现的次数。

样例：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-NKHJnKpf-1641003170440)(https://gitee.com/what_how_why2020/image-typora/raw/master/WordCount-mapreduce.png)]

理解以下代码的关键是明白Map和Reduce操作的输入数据、输出数据是什么

import java.io.IOException;
import java.util.StringTokenizer;
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;


public class WordCount {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        // 启动job任务
        Job job = Job.getInstance();
        job.setJobName("WordCount");
        // 设置mapper类、Reducer类
        job.setJarByClass(WordCount.class);  // 设置程序类
        job.setMapperClass(doMapper.class);  // 设置Mapper类
        job.setReducerClass(doReducer.class);  // 设置Reducer类

        // 设置Job输出结果的中key和value数据类型，因为结果是<单词,个数>，所以key设置为"Text"类型，Value设置为"IntWritable"类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // hdfs文件系统
        Path in = new Path("hdfs://localhost:9000/mr_demo/input/cust_fav"); //需要统计的文本所在位置
        Path out = new Path("hdfs://localhost:9000/mr_demo/output");  // 输出文件夹不能存在
        
        // 本地文件系统。若文件在本地文件系统，则替换为以下代码
        // Path in = new Path("file:///usr/local/java/data/mapreduce_demo/input/data_click"); // 用本地文件输入
		// Path out = new Path("file:///usr/local/java/data/mapreduce_demo/output"); // 结果输出到本地，文件夹不能已经存在
        
        // 设置job执行作业时输入和输出文件的路径
        FileInputFormat.addInputPath(job, in);
        FileOutputFormat.setOutputPath(job, out);
        
        // 无论程序是否执行成功，均强制退出
        // 如果程序成功运行，返回true，则程序返回0；如果程序执行失败，返回false，则程序返回1
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}



static class doMapper extends Mapper {

    public static final IntWritable one = new IntWritable(1);

    // 重写map方法，将k1,v1转为k2，v2
    @Override
    protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
        // 单词分隔符为t
        String[] splits = value.toString().split("t");
        for(String word: splits){
            context.write(new Text(word), one)
        }
    }
}

	

static class doReducer extends Reducer {
    private IntWritable result = 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();
        }
        result.set(sum);
        context.write(key, result);
    }
}

mapreduce中内置的WordCount程序位置：/usr/local/java/hadoop/hadoop-2.7.1/share/hadoop/mapreduce/hadoop-mapreduce-examples-2.7.1.jar

1. 将Hadoop的classhpath信息添加到CLASSPATH变量中，在终端中增加如下几行：

# 操作环境变量文件
sudo vim profile

# 写入以下两行数据
export HADOOP_HOME=/usr/local/hadoop 
export CLASSPATH=$($HADOOP_HOME/bin/hadoop classpath):$CLASSPATH

# 使环境变量生效
source profile

2. 在终端中使用javac命令编译文件

javac WordCount.java

3. 把.class文件打成jar包

jar -cvf WordCount.jar ./WordCount*.class

4. 运行程序，input和output文件均在/usr/local/java/data文件夹下

/usr/local/java/data jar WordCount.jar WordCount input output