mapreduce原理全剖析--段氏六脈神劍.png

1常熙、mapTask調(diào)用InputFormat再調(diào)用RecourReader的read()方法來讀取數(shù)據(jù)，獲得key墓贿、value，mapreduce通過InputFormat來解耦
2舱馅、read()方法依靠一次讀取一行的邏輯來讀取原始文件的數(shù)據(jù)，返回key干毅、value，mapTask會(huì)將其交給自定義的Mapper
3渠鸽、map方法我們會(huì)調(diào)用context.write方法來輸出數(shù)據(jù)到OutPutCollector類，OutPutCollector會(huì)將數(shù)據(jù)放到（內(nèi)存中存放 默認(rèn)MR.SORT.MB：100MB可以自己配置，一般不會(huì)放滿默認(rèn)80%那婉，這里面還要留有空間排序默認(rèn)20%）環(huán)形緩沖區(qū)(其實(shí)就是一個(gè)bite()數(shù)組，如果寫滿了，那么就會(huì)一邊寫一邊將開始的數(shù)據(jù)回收呻率，然后繼續(xù)寫到回收后的位置上逻悠，形成了環(huán)形緩沖區(qū))
4童谒、環(huán)形緩沖區(qū)的溢出的數(shù)據(jù)溢出之前會(huì)通過Hashpartioner進(jìn)行分區(qū)蔫饰、排序（默認(rèn)是快速排序法key.compareTO）,會(huì)通過spiller寫入到mapTask工作目錄的本地文件（所有溢寫文件分區(qū)且區(qū)內(nèi)有序）
5、所有溢出的文件會(huì)做歸并排序形成mapTask的最終結(jié)果文件，一個(gè)mapTask對(duì)應(yīng)一個(gè)最終結(jié)果文件盛嘿，形成幾個(gè)分區(qū)就會(huì)有對(duì)應(yīng)幾個(gè)reduceTask。reduceTask的個(gè)數(shù)由配置文件或者參數(shù)設(shè)置，只要不設(shè)置自定義partitioner，那么這里的分區(qū)會(huì)動(dòng)態(tài)適配reduceTask個(gè)數(shù)砰琢。如果設(shè)置了自定義partitioner，那么就需要提前設(shè)置對(duì)應(yīng)的reduceTask的個(gè)數(shù)
6、每個(gè)reduceTask都會(huì)到每一個(gè)mapTask的節(jié)點(diǎn)去下載分區(qū)文件到reduceTask的本地磁盤工作目錄
7歧强、為了保證最后的結(jié)果有序肤京，reduceTask任務(wù)A需要再次從所有mapTask下載到的對(duì)應(yīng)文件重新進(jìn)行歸并排序
8、reduceTask的內(nèi)部邏輯寫在reducer的reduce(key,values)方法，通過調(diào)用GroupingComparaor（key,netxtk）或者自定義GroupingComparaor來判斷哪些key是一組熬荆，形成key和values舟山。
9、reducer的reduce方法最后通過context.writer（key卤恳，v）寫到輸出文件(所有reduceTask的輸出文件都有序)累盗，輸出路徑由提交任務(wù)時(shí)的參數(shù)決定，默認(rèn)文件名part-r-00000
10突琳、如果設(shè)置了combiner若债，那么溢寫排序文件會(huì)調(diào)用，歸并排序時(shí)也會(huì)combiner拆融，將加快shluffer的效率蠢琳，但是一般情況下不建議使用,如果符合條件下一定要使用，也可以直接指定reducer為combiner镜豹，沒必要重復(fù)寫代碼

package cn.itcast.bigdata.mr.wcdemo;

import java.io.IOException;

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

/**
  * 輸入為map的輸出
 */
public class WordcountCombiner extends Reducer<Text, IntWritable, Text, IntWritable>{

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {

        int count=0;
        for(IntWritable v: values){
            
            count += v.get();
        }
        
        context.write(key, new IntWritable(count));
    }
}

***紅色方框的類和方法都可以由程序員自定義
mapper類

package cn.itcast.bigdata.mr.wcdemo;

import java.io.IOException;
import java.util.HashMap;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

/**
 * KEYIN: 默認(rèn)情況下傲须，是mr框架所讀到的一行文本的起始偏移量，Long,
 * 但是在hadoop中有自己的更精簡(jiǎn)的序列化接口趟脂，所以不直接用Long泰讽，而用LongWritable
 * 
 * VALUEIN:默認(rèn)情況下，是mr框架所讀到的一行文本的內(nèi)容昔期，String已卸，同上，用Text
 * 
 * KEYOUT：是用戶自定義邏輯處理完成之后輸出數(shù)據(jù)中的key硼一，在此處是單詞累澡，String，同上般贼，用Text
 * VALUEOUT：是用戶自定義邏輯處理完成之后輸出數(shù)據(jù)中的value永乌，在此處是單詞次數(shù)，Integer具伍，同上翅雏，用IntWritable
 * 
 * @author
 *
 */

public class WordcountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{

    /**
     * map階段的業(yè)務(wù)邏輯就寫在自定義的map()方法中
     * maptask會(huì)對(duì)每一行輸入數(shù)據(jù)調(diào)用一次我們自定義的map()方法
     */
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        
        //將maptask傳給我們的文本內(nèi)容先轉(zhuǎn)換成String
        String line = value.toString();
        //根據(jù)空格將這一行切分成單詞
        String[] words = line.split(" ");
        
        //將單詞輸出為<單詞，1>
        for(String word:words){
            //將單詞作為key人芽，將次數(shù)1作為value望几，以便于后續(xù)的數(shù)據(jù)分發(fā)，可以根據(jù)單詞分發(fā)萤厅，以便于相同單詞會(huì)到相同的reduce task
            context.write(new Text(word), new IntWritable(1));
        }
    }
}

reducer類

package cn.itcast.bigdata.mr.wcdemo;

import java.io.IOException;
import java.util.Iterator;

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

/**
 * KEYIN, VALUEIN 對(duì)應(yīng)  mapper輸出的KEYOUT,VALUEOUT類型對(duì)應(yīng)
 * 
 * KEYOUT, VALUEOUT 是自定義reduce邏輯處理結(jié)果的輸出數(shù)據(jù)類型
 * KEYOUT是單詞
 * VLAUEOUT是總次數(shù)
 * @author
 *
 */
public class WordcountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{

    /**
     * <angelababy,1><angelababy,1><angelababy,1><angelababy,1><angelababy,1>
     * <hello,1><hello,1><hello,1><hello,1><hello,1><hello,1>
     * <banana,1><banana,1><banana,1><banana,1><banana,1><banana,1>
     * 入?yún)ey橄抹，是一組相同單詞kv對(duì)的key
     */
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int count=0;
        /*Iterator<IntWritable> iterator = values.iterator();
        while(iterator.hasNext()){
            count += iterator.next().get();
        }*/
        for(IntWritable value:values){
        
            count += value.get();
        }
        context.write(key, new IntWritable(count));
    }
}

Driver類

package cn.itcast.bigdata.mr.wcdemo;

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.CombineTextInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 相當(dāng)于一個(gè)yarn集群的客戶端
 * 需要在此封裝我們的mr程序的相關(guān)運(yùn)行參數(shù)靴迫，指定jar包
 * 最后提交給yarn
 * @author
 *
 */
public class WordcountDriver {
    
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        
        //是否運(yùn)行為本地模式，就是看這個(gè)參數(shù)值是否為local楼誓，默認(rèn)就是local
        /*conf.set("mapreduce.framework.name", "local");*/
        
        //本地模式運(yùn)行mr程序時(shí)玉锌，輸入輸出的數(shù)據(jù)可以在本地，也可以在hdfs上
        //到底在哪里疟羹，就看以下兩行配置你用哪行主守，默認(rèn)就是file:///
        /*conf.set("fs.defaultFS", "hdfs://mini1:9000/");*/
        /*conf.set("fs.defaultFS", "file:///");*/
        
        
        
        //運(yùn)行集群模式，就是把程序提交到y(tǒng)arn中去運(yùn)行
        //要想運(yùn)行為集群模式榄融，以下3個(gè)參數(shù)要指定為集群上的值
        /*conf.set("mapreduce.framework.name", "yarn");
        conf.set("yarn.resourcemanager.hostname", "mini1");
        conf.set("fs.defaultFS", "hdfs://mini1:9000/");*/
        Job job = Job.getInstance(conf);
        
        job.setJar("c:/wc.jar");
        //指定本程序的jar包所在的本地路徑
        /*job.setJarByClass(WordcountDriver.class);*/
        
        //指定本業(yè)務(wù)job要使用的mapper/Reducer業(yè)務(wù)類
        job.setMapperClass(WordcountMapper.class);
        job.setReducerClass(WordcountReducer.class);
        
        //指定mapper輸出數(shù)據(jù)的kv類型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        
        //指定最終輸出的數(shù)據(jù)的kv類型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        
        //指定需要使用combiner参淫，以及用哪個(gè)類作為combiner的邏輯
        /*job.setCombinerClass(WordcountCombiner.class);*/
        job.setCombinerClass(WordcountReducer.class);
        
        //如果不設(shè)置InputFormat，它默認(rèn)用的是TextInputformat.class--CombineTextInputFormat是hdfs中要處理的文件都是大量小文件的情況下---主要是更改了getsplits()方法
        job.setInputFormatClass(CombineTextInputFormat.class);
        CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);//最大切片大小,如果超過最大值愧杯，會(huì)被切分開
        CombineTextInputFormat.setMinInputSplitSize(job, 2097152);//最小切片大小涎才，那么切下來實(shí)際上到底有多大？要考慮具體情況力九，本節(jié)點(diǎn)文件耍铜、跨機(jī)架文件、跨機(jī)房文件
        
        //指定job的輸入原始文件所在目錄
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        //指定job的輸出結(jié)果所在目錄
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        
        //將job中配置的相關(guān)參數(shù)跌前，以及job所用的java類所在的jar包业扒，提交給yarn去運(yùn)行
        /*job.submit();*/
        boolean res = job.waitForCompletion(true);
        System.exit(res?0:1);
        
    }
    

}