在windows系统上实现MapReduce的Word Count

这里使用的hadoop版本为：hadoop-2.6.0-cdh5.14.2.tar.gz
附带2个对应版本的文件：hadoop.dll
winutils.exe

一.配置环境

如上图所示进行配置。

二.将两个文件放到对应路径下

将hadoop.dll放到C:\Windows\System32目录下

将winutils.exe放到F:\hadoop-2.6.0-cdh5.14.2\bin目录下

三.创建maven工程

启动一个普通的maven工程，maven-quickstart即可。

1.先配置pom.xml

导入三个jar包

<dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>hadoop-common</artifactId>
  <version>2.6.0</version>
</dependency>

<dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>hadoop-hdfs</artifactId>
  <version>2.6.0</version>
</dependency>

<dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>hadoop-client</artifactId>
  <version>2.6.0</version>
</dependency>

2.创建一个Mapper类

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;

//创建Mapper类继承自泛型Mapper
/*
 *参数<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
 *KEYIN：表示mapper数据输入的时候key的数据类型，在默认的读取数据组件下，叫InputFormat，它的行为是一行一行的读取待处理的数据。读取一行，返回一行给程序，在这种情况下，KEYIN就表示每一行的起始偏移量，因此数据类型是Long，一般不改变它的类型
 *VALUEIN：表述mapper数据输入的的时候value的数据类型，在默认的读取数据组件下，VALUEIN就表示读取的这一行内容，因此数据类型是String
 *KEYOUT：表示mapper数据输出的时候key的数据类型，在本案例当中，输出的key是单词，因此数据类型是String
 *VALUEOUT表示mapper数据输出的视乎value的数据类型。在本案例当中，输出的key是单词的次数，因此数据类型是long
 *	long----LongWritable
 * String---Text
 * Integer--Intwritable
 * null-----NullWritable
 */
public class MyMapper extends Mapper<LongWritable,Text,Text,LongWritable>{
	//用于存放每个传过来的词频计数，为1
	private LongWritable one = new LongWritable(1);
	
	@Override
	//map方法对逐个<key,value>对进行处理
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    	//按照空格分割文本，装入数组
    	//Text 转String  word.toString()
        String[] wds = value.toString().split(" ");
        //遍历数组中的单词
        for (String word:wds){
        	//每有一个单词就标记1次
        	//String转 Text new Text(word) 
        	Text wd = new Text(word);
        	//通过context对象，将map的输出逐个输出
            context.write(wd,one);
        }
    }
}

3.创建一个Reducer类

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
/*Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
 *KEYIN：reducer阶段输入的数据key类型，对应mapper的输出key类型，这里就是Text
 *VALUEIN：reducer阶段输入的value类型，对应mapper的输出value类型，这里为单词次数LongWritable
 *KEYOUT：reducer阶段输出的数据key类型，这里为Text
 *VALUEOUT：reducer阶段输出的数据value类型，这里为单词的总次数LongWritable
 */
public class MyReduce extends Reducer<Text,LongWritable,Text,LongWritable>{
	private LongWritable res = new LongWritable();
	
	@Override
	protected void reduc(Text key,Iterable<LongWritable> values,Context context) throws IOException, InterruptedException{
		long resSum = 0;
		//依次获取单词数组中的每个单词，计数
		for(LongWritable one : values){
			resSum += one.get();		
		}
		//将统计完的数放到res LongWritable中，使可以序列化输出
		res.set(resSum);
		//将计算结果逐条输出
		context.write(key,res);
	}
}

4.创建driver类

import org.apache.hadoop.conf.Configuration;
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;


public class MyDriver {
    public static void main(String[] args) throws Exception {
        Configuration cfg =  new Configuration();
        //准备一个空任务
        Job job = Job.getInstance(cfg,"wc");
        //设置任务的输入数据源
        FileInputFormat.addInputPath(job,new Path("f:/md1.txt"));
        //设置你的Mapper任务类
        job.setMapperClass(MyMapper.class);
        //设置Mapper任务类的输出数据类型
        //有下面的setOutputKeyClass和setOutputValueClass，这两句可以不写
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);
        //设置你的Reducer任务类
        job.setReducerClass(MyReduce.class);
        //设置Reduce任务类输出的数据类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);
        //设置任务的输出数据目标
        FileOutputFormat.setOutputPath(job,new Path("f:/abc"));
        //启动任务并执行
        job.waitForCompletion(true);
    }
}

运行MyDriver，若如下图没看到报错，去目标文件目录下检查

找到输出的目标文件夹abc

找到文件

进行对比，看是否有误

以上为在Windows系统上实现MapReduce的Word Count