此文章已于 23:19:00 2015/8/4 重新发布到 农民阿姨
Hadoop之MapReduce
提交任务
hadoop jar hadoopdemo-1.0-SNAPSHOT-jar-with-dependencies.jar /user/root/testmr/hello.txt /user/root/testmr/out1
MapReduce原理***
MapReduce执行过程***
-------------------
hello you
hello me
-------------------
MapReducer的执行过程
1.map处理阶段
1.1读取HDFS的输入文件中的内容，把每一行解析成一个<k1,v1>对。
k1表示每一行的起始位置(单位byte)，v1表示每一行的文本内容。每个键值对调用一次map函数。
问：一共可以解析成多少个<k1,v1>?
答：2个，分别是<0, hello you>,<10, hello me>，一共2次map函数
1.2覆盖map函数，实现自己的业务逻辑。对输入的<k1,v1>做处理，转化为新的<k2,v2>输出。
void map(k1,v1, context){
String[] splited = v1.split('\t');
for(String word : splited){
context.write(<k2,v2>);
}
}
分别是<hello,1><you,1><hello,1><me,1>
1.3对map输出的<k2,v2>做分区。默认有1个分区。（分区的目的是让不同的reduce计算不同的数据）
1.4每个分区中的<k2,v2>按照k2进行排序、分组。分组指的是把相同k的v放到一个集合中。相同key的value放到一个集合中。
排序后<hello,1><hello,1><me,1><you,1>
分组后<hello,{1,1}><me,{1}><you,{1}>
1.5归约
2.reduce处理阶段
2.1对多个map的输出，按照不同的分区，通过网络copy到不同的reduce节点。
2.2对多个map的输出，进行合并、排序。覆盖reduce函数，实现自己的业务逻辑，对输入的<k2,v2s>进行处理，转化为新的<k3,v3>输出。
void reduce(k2,v2s, context){
long sum=0L;
for(long times : v2s){
sum += times;
}
context.write(k2,sum);
}
2.3把<k3,v3>写入到hdfs中
-------------------
问：从源代码的角度分析map函数处理的<k1,v1>是如何从HDFS文件中获取的？
答：
1.从TextInputFormat入手分析，找到父类FileInputFormat，找到父类InputFormat。
在InputFormat中找到2个方法，分别是getSplits(...)和createRecordReader(...)。
通过注释知道getSplits(...)作用是把输入文件集合中的所有内容解析成一个个的InputSplits，每一个InputSplit对应一个mapper task。
createRecordReader(...)作用是创建一个RecordReader的实现类。RecordReader作用是解析InputSplit产生一个个的<k,v>。
2.在FileInputFormat中找到getSplits(...)的实现。
通过实现，获知
(1)每个SplitSize的大小和默认的block大小一致，好处是满足数据本地性。
(2)每个输入文件都会产生一个InputSplit，即使是空白文件，也会产生InputSPlit；
如果一个文件非常大，那么会按照InputSplit大小，切分产生多个InputSplit。
3.在TextInputFormat中找到createRecordReader(...)的实现，在方法中找到了LineRecordReader。
接下来分析LineRecordReader类。
在RecordReader类中，通过查看多个方法，知晓key、value作为类的属性存在的，且知道了nextKeyValue()方法的用法。
在LineRecordReader类中，重点分析了nextKeyValue(...)方法。在这个方法中，重点分析了newSize = in.readLine(value, maxLineLength, maxBytesToConsume(pos));
在in.readLine(...)中，第一个形参存储被读取的行文本内容，返回值表示被读取内容的字节数。
通过以上代码，分析了InputSplit中的内容是如何转化为一个个的<k,v>。
4.从Mapper类中进行分析，发现了setup()、cleanup()、map()、run()。
在run()方法中，通过while，调用context.nextKeyValue(...)。
进一步分析Context的接口类是org.apache.hadoop.mapreduce.lib.map.WrappedMapper.MapContext，MapContext调用了nextKeyValue(...)。最终找到了MapContext的实现了MapContextImpl类org.apache.hadoop.mapreduce.task.MapContextImpl。
在这个类的构造方法中，发现传入了RecordReader的实现类。
package mapreduce;
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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
/**
* 假设有个目录结构
* /dir1
* /dir1/hello.txt
* /dir1/dir11/hello.bat
*
* 问：统计/dir1下面所有的文件中的单词技术
*
*/
public class WortCountTest {
public static void main(String[] args) throws Exception{
Configuration conf = new Configuration();
String jobName = WortCountTest.class.getSimpleName();
Job job = Job.getInstance(conf, jobName);
//要把代码打包运行， 调用如下行
job.setJarByClass(WortCountTest.class);
//指定输入路径
FileInputFormat.setInputPaths(job, new Path("hdfs://192.168.1.177:9000/wuchao/hello"));
//指定解析<k1,v1>的类
//job.setInputFormatClass(TextInputFormat.class);
//指定自定义的mapper类
job.setMapperClass(MyMapper.class);
//指定输出的<k2,v2>类型
//当<k3,v3>的类型与<k2,v2>类型一致时，<k2,v2>类型可以不指定
//job.setMapOutputKeyClass(Text.class);
//job.setMapOutputValueClass(LongWritable.class);
//指定自定义reducer类
job.setReducerClass(MyReducer.class);
//指定输出的<k3,v3>类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(LongWritable.class);
//指定输出<k3,v3>的类
//job.setOutputFormatClass(TextOutputFormat.class);
//指定输出的路径
FileOutputFormat.setOutputPath(job, new Path("hdfs://192.168.1.177:9000/out1"));
job.waitForCompletion(true);
}
private static class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable>{
Text k2 = new Text();
LongWritable v2 = new LongWritable();
@Override
protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, LongWritable>.Context context)
throws IOException, InterruptedException {
String line = value.toString();
String[] splited = line.split("\t");
//word表示每一行中的每个单词
for (String word : splited) {
k2.set(word);
v2.set(1L);
context.write(k2, v2);
}
}
}
private static class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable>{
LongWritable v3 = new LongWritable();
@Override
protected void reduce(Text k2, Iterable<LongWritable> v2s,
Reducer<Text, LongWritable, Text, LongWritable>.Context context) throws IOException, InterruptedException {
long sum = 0L;
for (LongWritable v2 : v2s) {
sum += v2.get();
}
v3.set(sum);
context.write(k2, v3);
}
}
}
jar 包提交
hadoop jar /home/lina/example.jar com/oss/maxtemperature/MaxTemperatureDriver /local/sample.txt /output/sampleout