码迷,mamicode.com
首页 > 其他好文 > 详细

Hadoop Mapreduce 工作机制

时间:2019-03-11 01:02:11      阅读:205      评论:0      收藏:0      [点我收藏+]

标签:input   图片   mapr   定义   throw   代码   产生   void   percent   

一.Mapreduce 中的Combiner

技术图片

技术图片

 

 

 

package com.gec.demo;


import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class WcCombiner extends Reducer<Text, IntWritable,Text,IntWritable> {
    private IntWritable sum=new IntWritable();
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int count=0;

        for (IntWritable value : values) {
            count+=value.get();
        }
        sum.set(count);
        context.write(key,sum);
    }
}

在job类中声明如下:

技术图片

技术图片

 

二.MapTask工作机制

主要的核心类:

读:
FileInputFormat
                   TextInputFormat
                               createRecordReader
                                                LineRecordReader
                                                                      nextKeyValue

写:
context.write
            RecordWriter.write(k,value)
                            NewOutputCollector.write(key,value)
                                                    MapOutputCollector.collect(key,value,partitions)
                                                                                            MapOutputBuffer.collect(key,value,partitions)

技术图片

 

核心map输出源代码分析类
NewOutputCollector类
构造器:
实例化MapOutputBuffer对象
调用MapOutputBuffer对象init方法
将MapOutputBuffer对象赋值给collector对象
解决分区值问题
//如果没有自定义分区类,则默认使用HashPartitioner
partitioner = (org.apache.hadoop.mapreduce.Partitioner<K,V>)
ReflectionUtils.newInstance(jobContext.getPartitionerClass(), job);

环形缓冲区实现原理

MapOutputBuffer实现缓冲区的核心实现

技术图片

在这一头存储key和value,key和value依次排列,而那一头存储索引,向中间出发,当储存的空间占比百分之八十的时候,则溢出,两者的方向改变,分别开始从另外一头开始存储

 技术图片

如上图,从赤道分别向不同方向出发

技术图片

如上图,到达溢出时,产生新赤道,又分别从新赤道往回走

 

 

 

init方法
1、分配溢出比
final float spillper =
job.getFloat(JobContext.MAP_SORT_SPILL_PERCENT, (float)0.8);
2、分配环形缓存区的大小
final int sortmb = job.getInt(JobContext.IO_SORT_MB, 100);
3、实例化快排对象
sorter = ReflectionUtils.newInstance(job.getClass("map.sort.class",
QuickSort.class, IndexedSorter.class), job);
4、定义环形缓存区数组
kvmeta = ByteBuffer.wrap(kvbuffer)
.order(ByteOrder.nativeOrder())
.asIntBuffer();
5、开始化赤道
setEquator(0);
6、获取key的比较器对象
comparator = job.getOutputKeyComparator();

7、是否定义combineCollector对象

8、启动spillThread线程,监听溢出比,触发此 sortAndSpill()

技术图片

技术图片

 

 


对接Reducer类的方法:
reduce(Text key, Iterable<IntWritable> values, Context context)

ReduceTask工作机制

技术图片

 

三.shuffer

shuffer缓存流程

技术图片

技术图片

 

Hadoop Mapreduce 工作机制

标签:input   图片   mapr   定义   throw   代码   产生   void   percent   

原文地址:https://www.cnblogs.com/Transkai/p/10508165.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!