Mapreduce简介
Hadoop MapReduce 源于Google发表的 MapReduce论文。Hadoop MapReduce 其实就是Google MapReduce的一个克隆版本。Hadoop 2.0即第二代Hadoop系统,其框架最核心的设计是HDFS、MapReduce和YARN。其中,HDFS为海量数据提供存储,MapReduce用于分布式计算,YARN用于进行资源管理。
MapReduce的工作流程图
其实,一次mapreduce过程就包括上图的6个步骤,input、splitting、mapping、shuffling、redecing、final redult。
-
Input
文件要存储在HDFS中,每个文件被切分成多个一定大小的块也就是Block,(Hadoop1.0默认为64M,Hadoop2.0默认为128M),并且默认3个备份存储在多个的节点中。
MR通过Inputformat将数据文件从HDFS中读入取,读取完后会对数据进行split切片,切片的数量根据Block的大小所决定,然后每一个split的个数又决定map的个数,即一个split会分配一个maptask并行实例处理。
如何确定切分的文件大小?
默认split的大小与block的大小相同,为128MB。
split大小由minSize、maxSize、blockSize决定
protected long computeSplitSize(long blockSize, long minSize, long maxSize) {
return Math.max(minSize, Math.min(maxSize, blockSize));
}
-
Mapper
数据进入到map函数中,然后开始按照一定的规则切分。其实这就是我们自定义的计算逻辑,我们编写mr程序的map函数的逻辑一般就在这个阶段执行。企业应用为了追求开发效率,一般都使用hive sql代替繁琐的mr程序了,这里附上一个经典的wordcount的map函数重温一下吧。
// Mapper四个参数:第一个Object表示输入key的类型;第二个Text表示输入value的类型;第三个Text表示表示输出键的类型;第四个IntWritable表示输出值的类型。map这里的输出是指输出到reduce
public static class doMapper extends Mapper<Object, Text, Text, IntWritable> {
public static final IntWritable one = new IntWritable(1);//这里的IntWritable相当于Int类型
public static Text word = new Text();//Text相当于String类型
// map参数<keyIn key,valueIn value,Context context>,将处理后的数据写入context并传给reduce
protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
//StringTokenizer是Java工具包中的一个类,用于将字符串进行拆分
StringTokenizer tokenizer = new StringTokenizer(value.toString(), " ");
//返回当前位置到下一个分隔符之间的字符串
word.set(tokenizer.nextToken());
//将word存到容器中,记一个数
context.write(word, one);
}
}
-
shuffle
Shuffle是我们不需要编写的模块,但却是十分关键的模块。
在map中,每个 map 函数会输出一组 key/value对, Shuffle 阶段需要从所有 map主机上把相同的 key 的 key value对组合在一起,(也就是这里省去的Combiner阶段)组合后传给 reduce主机, 作为输入进入 reduce函数里。
Partitioner组件 负责计算哪些 key 应当被放到同一个 reduce 里
HashPartitioner类,它会把 key 放进一个 hash函数里,然后得到结果。如果两个 key 的哈希值 一样,他们的 key/value对 就被放到同一个 reduce 函数里。我们也把分配到同一个 reduce函数里的 key /value对 叫做一个reduce partition.
我们看到 hash 函数最终产生多少不同的结果, 这个 Hadoop job 就会有多少个 reduce partition/reduce 函数,这些 reduce函数最终被JobTracker 分配到负责 reduce 的主机上,进行处理。
Map方法之后,数据首先进入到分区方法,把数据标记好分区,然后把数据发送到环形缓冲区;环形缓冲区默认大小100m,环形缓冲区达到80%时,进行溢写;溢写前对数据进行排序,排序按照对key的索引进行字典顺序排序,排序的手段快排;溢写产生大量溢写文件,需要对溢写文件进行归并排序;对溢写的文件也可以进行Combiner操作,前提是汇总操作,求平均值不行。最后将文件按照分区存储到磁盘,等待Reduce端拉取。每个Reduce拉取Map端对应分区的数据。拉取数据后先存储到内存中,内存不够了,再存储到磁盘。拉取完所有数据后,采用归并排序将内存和磁盘中的数据都进行排序。在进入Reduce方法前,可以对数据进行分组操作。值得注意的是,整个shuffle操作是有3次排序的。
同时reduce任务并不是在map任务完全结束后才开始的,Map 任务有可能在不同时间结束,所以 reduce 任务没必要等所有 map任务 都结束才开始。事实上,每个 reduce任务有一些 threads 专门负责从 map主机复制 map 输出(默认是5个)。
-
Reduce
reduce() 函数以 key 及对应的 value 列表作为输入,按照用户自己的程序逻辑,经合并 key 相同的 value 值后,产 生另外一系列 key/value 对作为最终输出写入 HDFS。