MapReduce简介，结构组成，运行过程，WordCount...

yarn        //resourcemanager + nodemanager

1.MapReduce简介

Hadoop最主要的两部分Hdfs,MapReduce。其中Hdfs是用来分布式存储数据。在面对着由Hdfs存储的巨大数据量时，设计面向大数据并行处理的计算模型势在必得。

  MapReduce最早是由Google公司研究提出的一种面向大规模数据处理的并行计算模型和方法。Google公司设计MapReduce的初衷主要是为了解决其搜索引擎中大规模网页数据的并行化处理。Google公司发明了MapReduce之后首先用其重新改写了其搜索引擎中的Web文档索引处理系统。但由于MapReduce可以普遍应用于很多大规模数据的计算问题，因此自发明MapReduce以后，Google公司内部进一步将其广泛应用于很多大规模数据处理问题。到目前为止，Google公司内有上万个各种不同的算法问题和程序都使用MapReduce进行处理。

2.MapReduce结构组成

MapReduce由两个阶段组成：Map阶段和Reducer阶段。用户只需要实现map()和reduce()两个函数，即可实现分布式计算，非常简单。其中这两个函数的形参是key,value对，表示函数的输入信息。

Map与Reducer各有两对key,value对，分别是输入key,value对，以及输出key,value对。

其中Map的输入key,value对是读取数据存储之后生成的，输出key,value对则是对读取的数据进行处理产生的。

Reducer的输入key,value对是Map的输出key,value对，输出key,value对则是最终要的结果。

3.MapReduce运行过程

在面对巨大的数据存储时，如何来设计计算模型呢？MapReduce计算模型采用的是分而治之的原理。如下图所示：

对于求1+5+7+3+4+9+3+5+6的时候，MapReduce使用切片的形式，将数据切分成3（1+5+7,3+4+9,3+5=6）份，然后分别在不同的Map节点上同时进行运算，这样对大数据处理而言速度效率机会快很多，最后把各个节点上算出的数据13，16,14在Reduce进行最后的整合，得出最终的结果。

那么数据在Map端与Reducer端是怎样传输的，以及传输过程中都发生了些什么呢？让我们来看一下：

假设数据如上图所示。

第一步，对于存储在Hdfs上的数据，首先会使用InputFormat——一个抽象类，其实现类包括文本文件实现类：TextInputFormat(默认输入实现类)，序列文件输入类：SequenceFileInputFormat等。进行输入。

第二步，为了提高运算效率，对巨大数据进行数据切片，并行计算。切片的计算方法是对于一个文件（最小切片minSize大小是1，最大切片minSplit大小是Long.Max，数据块大小blockSize）取3者中的中间值。

第三步，进行数据切割之后获取切片的数量后开启Map过程，有多少个切片，就有多少个Map。Map过程对输入的<K,V>对进行处理，然后输出<K,V>对。

第四步，在Map阶段，对输出的<K,V>对进行分区，目的是把当前输出的<K,V>划分到哪一个Reducer中，默认的分区函数是对K进行Hash。

第五步，分区之后的<K,V>对会保存到内存缓冲区，当数据量达到缓冲区的百分之80之后，缓冲区会开启一个线程，将数据写入到磁盘的一个临时文件中，在写入之前，对分区内的数据进行排序，排序的原则是按照字典顺序进行排序，并且把相同的Key的进行归并，例如把<hello,1>,<hello,2>,<hello,1>归并成<hello,{1,2,1}>

第六步，多个Map任务的输出，将进行排序并且归并。例如Map1中的1号分区<hello,{1,2,1}>与Map2中1号分区<hello,{1,1,1}>归并成<hello,{1,2,1,1,1,1}>。

第七步，每个Reducer（默认是1个，也可以自行进行配置）通过网络的拷贝，将自己对应的分区数据拷贝到自己的节点，然后进行业务逻辑处理，最后输出。

4.WordCount实例

根据上面的流程分析，我们来写一个经典的MapReduce案例——WordCount，统计文件中的单词数量。不要小看这个案例，百度中的热词搜索与其原理有点类似，统计出每日搜索最多的词。

数据文件内容如下

Map类

package com.hadoop.mr;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**

* @author zpx

* @Title: ${file_name}

* @Package ${package_name}

* @Description: WordCount的Mapper

* @date 2018/6/2811:13

*/

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

    @Override

    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        Text keyOut=new Text();

        IntWritable valueOut=new IntWritable();

        String[] arr = value.toString().split(" ");

        for(String s:arr){

           keyOut.set(s);

           valueOut.set(1);

            context.write(keyOut,valueOut);

        }

    }

}

Reduce类

package com.hadoop.mr;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/**

* @author zpx

* @Title: ${file_name}

* @Package ${package_name}

* @Description: WordCount的Reduce

* @date 2018/6/2811:24

*/

public class WCReducer extends Reducer<Text,IntWritable,Text,IntWritable>{

    @Override

    /**

    　　* @Description:一个分组只进入一次reduce

    　　* @param ${tags}

    　　* @return ${return_type}

    　　* @throws

    　　* @author zpx

    　　* @date 2018/6/30 0:15

    　　*/

    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));

    }

}

主函数类

package com.hadoop.mr;

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.FileInputFormat;

import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**

* @author zpx

* @Title: ${file_name}

* @Package ${package_name}

* @Description: WordCount主函数

* @date 2018/6/2811:29

*/

public class WCApp{

    public static void main(String[] args) throws Exception {

        //在程序中指定hadoop.home.dir

//        System.setProperty("hadoop.home.dir","F://hadoop-dir" );

        Configuration conf=new Configuration();

        Job job=Job.getInstance(conf);

//        conf.set("fs.defaultFS", "file:///");//本地测试请将注释取消掉

        //设置job的各种属性

        job.setJobName("WCApp");                        //作业名称

        job.setJarByClass(WCApp.class);                 //搜索类

        job.setInputFormatClass(TextInputFormat.class); //设置输入文件格式

        //添加输入路径

        FileInputFormat.addInputPath(job,new Path(args[0]));

        //设置输出路径

        FileOutputFormat.setOutputPath(job,new Path(args[1]));

        //设置最大切片数

        //FileInputFormat.setMaxInputSplitSize(job,13);

        //最小切片数

        //FileInputFormat.setMinInputSplitSize(job,1L);

        //设置分区类

//        job.setPartitionerClass(MyPartitioner.class);   //设置自定义分区

        //设置合成类

//        job.setCombinerClass(WCReducer.class);          //设置combiner类

        job.setMapperClass(WCMapper.class);             //mapper类

        job.setReducerClass(WCReducer.class);           //reducer类

        //job.setNumReduceTasks(3);                       //reduce个数

        //设置Map的输入输出

        job.setMapOutputKeyClass(Text.class);           //

        job.setMapOutputValueClass(IntWritable.class);  //

        //设置Reducer的输入数据

        job.setOutputKeyClass(Text.class);

        job.setOutputValueClass(IntWritable.class);     //

        job.waitForCompletion(true);

    }

将当前工程打成JAR包，然后放在集群上进行测试，命令：

hadoop jar mr.jar com.hadoop.mr.WCApp hdfs://s10/user/zpx/wc/data hdfs://s10/user/zpx/wc/out

查看结果

过程分析：使用MapReduce默认的数据输入格式TextInputFormat.class，对数据切片读取的过程中，输入Key是文件的偏移量（即每一行开始位置）,Value是一整行数据，所以在Map类中要对数据进行切割，然后输出Key,Value。格式是<hello,1>等。进行了排序归并之后输入到Reduce端的K,V对格式是<hello,{1,1,1,1,1,1,1,1,1,1,1,1,1,1}>,<tom,{1,1,1,1,1,1,1}>,<tom0,{1}>,<tom0,{1}>,<tom1,{1}>,<tom2,{1,1,1}>,<tom3,{1}>,<tom4,{1}>,Reduce端只需要对相应Key的数据集合做迭代运算就可以了。最后生成Key,Value对输出即可。由于在此过程中要数据要在网络中传输，所以要对数据进行序列化，而Hadoop有自己的一套序列化标准（未使用Java的序列化，因为其麻烦，复杂），所以我们看到的Key,Value是Hadoop自己的序列化，Long->LongWritable,Int->IntWritable,String->Text