本文讲解利用MR实现简单的单词统计功能。
创建com.test.hadoop.wordcount包，然后在包中创建自定义mapper类，自定义reducer类，以及main类。
Mapper代码
package com.kaikeba.hadoop.wordcount;
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;
/**
* 类Mapper<LongWritable, Text, Text, IntWritable>的四个泛型分别表示
* map方法的输入的键的类型kin、值的类型vin；输出的键的类型kout、输出的值的类型vout
* kin指的是当前所读行行首相对于split分片开头的字节偏移量,所以是long类型，对应序列化类型LongWritable
* vin指的是当前所读行，类型是String，对应序列化类型Text
* kout根据需求，输出键指的是单词，类型是String，对应序列化类型是Text
* vout根据需求，输出值指的是单词的个数，1，类型是int，对应序列化类型是IntWritable
*
*/
public class WordCountMap extends Mapper<LongWritable, Text, Text, IntWritable> {
/**
* 处理分片split中的每一行的数据；针对每行数据，会调用一次map方法
* 在一次map方法调用时，从一行数据中，获得一个个单词word，再将每个单词word变成键值对形式(word, 1)输出出去
* 输出的值最终写到本地磁盘中
* @param key 当前所读行行首相对于split分片开头的字节偏移量
* @param value 当前所读行
* @param context
* @throws IOException
* @throws InterruptedException
*/
public void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
//当前行的示例数据(单词间空格分割)：Dear Bear River
//取得当前行的数据
String line = value.toString();
//按照\t进行分割，得到当前行所有单词
String[] words = line.split("\t");
for (String word : words) {
//将每个单词word变成键值对形式(word, 1)输出出去
//同样，输出前，要将kout, vout包装成对应的可序列化类型，如String对应Text，int对应IntWritable
context.write(new Text(word), new IntWritable(1));
}
}
}
Reducer代码
package com.kaikeba.hadoop.wordcount;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
/**
*
* Reducer<Text, IntWritable, Text, IntWritable>的四个泛型分别表示
* reduce方法的输入的键的类型kin、输入值的类型vin；输出的键的类型kout、输出的值的类型vout
* 注意：因为map的输出作为reduce的输入，所以此处的kin、vin类型分别与map的输出的键类型、值类型相同
* kout根据需求，输出键指的是单词，类型是String，对应序列化类型是Text
* vout根据需求，输出值指的是每个单词的总个数，类型是int，对应序列化类型是IntWritable
*
*/
public class WordCountReduce extends Reducer<Text, IntWritable, Text, IntWritable> {
/**
*
* key相同的一组kv对，会调用一次reduce方法
* 如reduce task汇聚了众多的键值对，有key是hello的键值对，也有key是spark的键值对，如下
* (hello, 1)
* (hello, 1)
* (hello, 1)
* (hello, 1)
* ...
* (spark, 1)
* (spark, 1)
* (spark, 1)
*
* 其中，key是hello的键值对被分成一组；merge成[hello, Iterable(1,1,1,1)]，调用一次reduce方法
* 同样，key是spark的键值对被分成一组；merge成[spark, Iterable(1,1,1)]，再调用一次reduce方法
*
* @param key 当前组的key
* @param values 当前组中，所有value组成的可迭代集和
* @param context reduce上下文环境对象
* @throws IOException
* @throws InterruptedException
*/
public void reduce(Text key, Iterable<IntWritable> values,
Context context) throws IOException, InterruptedException {
//定义变量，用于累计当前单词出现的次数
int sum = 0;
for (IntWritable count : values) {
//从count中获得值，累加到sum中
sum += count.get();
}
//将单词、单词次数，分别作为键值对，输出
context.write(key, new IntWritable(sum));// 输出最终结果
};
}
Main程序入口
package com.kaikeba.hadoop.wordcount;
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;
import java.io.IOException;
/**
*
* MapReduce程序入口
* 注意：
* 导包时，不要导错了；
* 另外，map\reduce相关的类，使用mapreduce包下的，是新API，如org.apache.hadoop.mapreduce.Job;；
*/
public class WordCountMain {
//若在IDEA中本地执行MR程序，需要将mapred-site.xml中的mapreduce.framework.name值修改成local
//参数 c:/test/README.txt c:/test/wc
public static void main(String[] args) throws IOException,
ClassNotFoundException, InterruptedException {
//判断一下，输入参数是否是两个，分别表示输入路径、输出路径
if (args.length != 2 || args == null) {
System.out.println("please input Path!");
System.exit(0);
}
Configuration configuration = new Configuration();
//configuration.set("mapreduce.framework.name","local");
//告诉程序，要运行的jar包在哪
//configuration.set("mapreduce.job.jar","/home/hadoop/IdeaProjects/Hadoop/target/com.kaikeba.hadoop-1.0-SNAPSHOT.jar");
//调用getInstance方法，生成job实例
Job job = Job.getInstance(configuration, WordCountMain.class.getSimpleName());
//设置job的jar包，如果参数指定的类包含在一个jar包中，则此jar包作为job的jar包； 参数class跟主类在一个工程即可；一般设置成主类
// job.setJarByClass(WordCountMain.class);
job.setJarByClass(WordCountMain.class);
//通过job设置输入/输出格式
//MR的默认输入格式是TextInputFormat，输出格式是TextOutputFormat；所以下两行可以注释掉
// job.setInputFormatClass(TextInputFormat.class);
// job.setOutputFormatClass(TextOutputFormat.class);
//设置输入/输出路径
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
//设置处理Map阶段的自定义的类
job.setMapperClass(WordCountMap.class);
//设置map combine类，减少网路传出量
job.setCombinerClass(WordCountReduce.class);
//设置处理Reduce阶段的自定义的类
job.setReducerClass(WordCountReduce.class);
//注意：如果map、reduce的输出的kv对类型一致，直接设置reduce的输出的kv对就行；如果不一样，需要分别设置map, reduce的输出的kv类型
//注意：此处设置的map输出的key/value类型，一定要与自定义map类输出的kv对类型一致；否则程序运行报错
// job.setMapOutputKeyClass(Text.class);
// job.setMapOutputValueClass(IntWritable.class);
//设置reduce task最终输出key/value的类型
//注意：此处设置的reduce输出的key/value类型，一定要与自定义reduce类输出的kv对类型一致；否则程序运行报错
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
// 提交作业
job.waitForCompletion(true);
}
}
本地运行
初次运行WordCountMain，先设置main方法参数，根据图示操作即可
设置参数
设置参数
设置参数
然后在Program arguments输出参数：c:/test/README.txt c:/test/wc（两个参数间有一个英文空格，表示两个参数）
在WordCountMain代码上，点击鼠标右键，运行程序。
运行
网友评论