Hadoop学习日记(七)
Hadoop学习日记(七)
2019.03.08
主题:wordcount编写和提交集群运行
这部分内容的学习主要参考段海涛Hadoop之p24
这部分的学习主要是编写了WCMapper、WCReducer、WCRunner三个类,以及最终打包成jar包提交集群运行。
一、关键点汇总
1. 在wordcount实例里,WCReducer中Reduce()方法数据输入的value其实是一个list
2.FileInputFormat以及FileOutputFormat应从包org.apache.hadoop.mapreduce.lib.input和org.apache.hadoop.mapreduce.lib.output中导入,Hadoop老的接口不要使用
3. 最终wordcount实例运行的结果如下图所示,可以看到结果按照key进行排序
4. 另外还有一些注意点标注在源码的注释里。
二、源码
1.WCMapper
package weekend111;
import java.io.IOException;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
//四个泛型中,前两个是指定mapper输入数据的类型,KEYIN是输入的key的类型VALUEIN时输入的value的类型
//map和reduce的数据输入输出都是以key-value对的形式封装的
//默认情况下,框架传递给我们的mapper的输入数据中,key是要处理的文本中一行的起始偏移量,这一行的内容作为value
public class WCMapper extends Mapper<LongWritable, Text, Text, LongWritable>{
//mapreduce框架每读一行数据就调用一次该方法
@Override
protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, LongWritable>.Context context)
throws IOException, InterruptedException {
// TODO Auto-generated method stub
//具体业务逻辑就写在这个方法中,而且我们业务要处理的数据已经被框架传递进来,在方法的参数key-value中
//key是这一行数据的起始偏移量 value是这一行的文本内容
//super.map(key, value, context);
String line=value.toString();
String[] words=StringUtils.split(line," ");
//遍历这个单词数组输出为kv形式 k:单词 v:1
for(String word:words)
{
context.write(new Text(word), new LongWritable(1));
}
}
}
2.WCReducer
package weekend111;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
public class WCReducer extends Reducer<Text, LongWritable, Text, LongWritable>{
//框架在map处理完成后,将所有kv对缓存起来,进行分组,然后传递一个组<key,values{}>,调用一次reduce方法
//<hello,{1,1,1,1,1......}>
@Override
protected void reduce(Text arg0, Iterable<LongWritable> arg1,
Reducer<Text, LongWritable, Text, LongWritable>.Context arg2) throws IOException, InterruptedException {
// TODO Auto-generated method stub
//super.reduce(arg0, arg1, arg2);
long count=0;
//遍历value的list,进行累加求和
for(LongWritable value:arg1)
{
count+=value.get();
}
//输出这一个单词的统计结果
arg2.write(arg0, new LongWritable(count));
}
}
3.WCRunner
package weekend111;
import java.io.IOException;
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;
/**
* 用来描述一个特定的作业
* 比如,改作业使用哪个类作为逻辑处理中的map,哪个作为reduce
* 还可以指定该作业要处理的数据所在的路径
* 还可以指定改作业输出的结果放到哪个路径
* @author Administrator
*
*/
public class WCRunner {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
Configuration conf=new Configuration();
Job job=Job.getInstance(conf);
//设置整个job所用的那些类在哪个jar包
job.setJarByClass(WCRunner.class);
//本job使用的的mapper和reducer的类
job.setMapperClass(WCMapper.class);
job.setReducerClass(WCReducer.class);
//指定reduce的输出数据kv类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(LongWritable.class);
//指定mapper的输出数据类型
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(LongWritable.class);
//指定要处理的输入数据存放路径
FileInputFormat.setInputPaths(job,new Path("/wc/srcdata/"));
//指定处理结果的输出数据的存放路径
FileOutputFormat.setOutputPath(job, new Path("/wc/output/"));
//将job提交给集群运行
job.waitForCompletion(true);
}
}