【大数据系列】hadoop核心组件-MapReduce

一、引入

hadoop的分布式计算框架（MapReduce是离线计算框架）

二、MapReduce设计理念

移动计算，而不是移动数据。

Input HDFS先进行处理切成数据块（split）   map   sort   reduce  输出数据（output HDFS）

三、示例

Mapping是根据我们书写的模式执行的。

四、hadoop计算框架Shuffle

怎样将map task的输出结果有效的传送到reduce端，也就是说，Shuffle描述着数据从map task输出到reduce task输入的这段过程

1、在mapper和reducer中间的一个步骤

2、可以把mapper的输出按照某种key值重新切分和组合成n份，把key值符合某种范围的输出送到特定的reducer那里去处理。

3、 可以简化reducer过程

 

Partition的作用是根据key或value及reduce的数量来觉得当前的这对输出最终应该交由哪个reduce task处理。默认对key hash后再以reduce task数量取模。默认的取模方式只是为了平均reduce的处理能力，如果用户自己对Partitioner有需求，可以订制并设置到job上。

4、 每个map task都有一个内存缓冲区（默认是100MB），存储着map的输出结果。

当缓存区快满的时候需要将缓冲区的数据以一个临时文件的方式存放到磁盘（spill）

溢写是由单独线程来完成，不影响往缓冲区写map的线程，溢写线程启动时不应该阻止map的结果输出，所以整个缓冲区有个溢写比的比例（spill.percent，默认是0.8），假如是100MB，溢写线程启动的时候锁定这80MB的内存，执行溢写过程，Map task的输出结果还可以往剩下的20MB内存中写，互不影响。

当溢写线程启动后，需要对这80MB空间的key做排序。

        （Map端执行过程）

假如Client设置过Combiner,那么现在就是使用Combiner的时候了。将有相同key的key/value对的value加起来。减少溢写到磁盘的数据量（reducer1,word1,[9]）

当整个map task结束后再对次哦按中这个map task产生的所有临时文件做合并（Merge）,对于“word1”就是这样的：{“word1”,[5,8,2,…]},假如有Combiner,{word1[15],}最终产生一个文件。

Reduce从tasktracker copy数据

Copy过来的数据会先放入内存缓冲区中，这里的缓冲区大小要比map端的更为灵活，它基于JVM的heap size设置

Merge有三种形式：1）内存到内存 2）内存到磁盘 3）磁盘到磁盘。Merge从不同的tasktracker上拿到的数据，{word1[15,17,2]}

Combiner会优化MapReduce的中间结果，Combiner的输出是Reducer的输入，Combiner绝不能改变最终的计算结果。所以Combiner只应该用于那种Reduce的输入key/value与输出key/value类型完全一致，且不影响最终结果的场景。

                                                   （reducer端）

Copy过程，简单的拉取数据，Reduce进程启动一些数据copy线程（Fetcher）,通过HTTP方式请求map task所在的TaskTracker获取map task的输出文件。因为map task早已结束，这些文件就归TaskTracker管理在本地磁盘中。

Merge阶段，这里的merge如map端的merge动作，只是数组中存放的是不同map端copy来的数值。Copy过来的数据会先放入内存缓冲区中，这里的缓冲区大小要比map端的更为灵活，它基于JVM的heap size来设置，因为Shuffle阶段Reduce不允许，所以应该把大部分的内存都给Shuffle用。

Reducer的输入文件，不断的merge后，最终会生成一个“最终文件”，这个文件可能在磁盘上，也可能在内存上，直接作为Rducer的输入，当Rueducer的输入文件已定，整个Shuffle才最终结束，然后就是Reducer执行，把结果放到HDFS上。

五、MapReduce的Split大小：

     Max.split:100MB

       Min.split:10MB

       Block:64MB

       Max(min.split,min(max.split,block))

六、MapReduce的架构

1、 一主多从架构

2、主JobTracker

   负责调度分配每一个子任务task运行在TaskTracker上，如果发现有失败的etask就重新分配其任务到其他节点，每一个hadoop集群中只一个JobTracker一般它运行在Master节点上

3、从TaskTracker

TaskTracker主动与JobTracker通信，接收作业，并负责直接执行每一个任务，为了减少网络带宽TaskTracker最好运行在HDFS的DataNode上

七、MapReduce安装

1．选择一台作为JobTracker(例如node3)

2． 编辑mapred-site.xml(NameNode机器上的)

3、将刚才编辑的文件同步到别的机器上

4、 启动（star-all.sh）如果没有安装mapreduce的时候启动使用hdfs-start.sh

5、访问http://node3.50030

6、编写测试wordCount类

package org.slp;

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

import java.io.IOException;
import java.util.StringTokenizer;

/**
 * Created by sanglp on 2017/7/16.
 */
public class WcMapper extends Mapper<LongWritable,Text,Text,IntWritable>{
    /**
     * 每次调用map方法会传入split中一行数据key:该行数据所在文件中的位置下标。value:这行行行数据
     * map会调用多次的
     * @param key
     * @param value
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //super.map(key, value, context);
        String line = value.toString();
        StringTokenizer st = new StringTokenizer(line);//输入以空格切开
        while (st.hasMoreTokens()){
            String world = st.nextToken();
            context.write(new Text(world),new IntWritable(1));//map的输出
        }
    }

}

package org.slp;

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

import java.io.IOException;

/**
 * Created by sanglp on 2017/7/16.
 */
public class WcReducer extends Reducer<Text,IntWritable,Text,IntWritable> {
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        //super.reduce(key, values, context);
        int sum = 0 ;
        for(IntWritable i : values){
            sum = sum+i.get();
        }
        context.write(key,new IntWritable(sum));
    }
}

package org.slp;

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.output.FileOutputFormat;

import java.io.IOException;

/**
 * Created by sanglp on 2017/7/16.
 */
public class JobRun {
    public static void main(String[] args){
        Configuration conf = new Configuration();
        conf.set("mapred.job.tracker","node1:9001");
        try {
            Job job = new Job(conf);
            job.setJarByClass(JobRun.class);
            job.setMapperClass(WcMapper.class);
            job.setReducerClass(WcReducer.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(IntWritable.class);

            job.setNumReduceTasks(1);//设置reduce任务的个数
            //mapreduce输入数据所在目录或文件
            FileInputFormat.addInputPath(job,new Path("/usr/input/wc"));
            //mr执行之后的输出数据的目录
            FileOutputFormat.setOutputPath(job,new Path("/usr/out/wc"));
            try {
                System.exit(job.waitForCompletion(true)?0:1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ClassNotFoundException e) {
                e.printStackTrace();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

7、将编写的文件导出为jar包命名为wc.jar

8、./hadoop jar /root/wc.jar com.slp.mr.JobRun