大数据Hadoop基础MapReduce详解

MapReduce原理篇

MapReduce是一个分布式运算框架，主要的功能是将用户编写的业务逻辑代码和自带默认的组件整合在一起形成一个完整的分布式运算程序，并发的运行在一个集群上。

客户端提交MapReduce任务的过程

流程如下：

1.用户提交job任务之后，程序运行job.sumbit()方法，这是MapReduce框架会使用JobSubmitter的成员变量Cluster，Cluster对象会使用内部的Proxy对象，如果客户端提交的是云端运行，那么使用YarnRunner，如果提交的是本地运行，使用的是LocalRunner。

2.如果提交的是集群运行，那么将会返回一个HDFS文件目录，路径的形式是hdfs://..../.staging，对于每一个MR程序，MapReduce会分配一个jobID，程序根据这两个值拼接称为一个新的路径。

3.MapReduce程序调用用户指定FileInputFormat的实现类的getSplits()方法获取切片来进行mapTask并行度，这里形成job.split文件、以及job相关参数形成的文件job.xml和job的jar包。

4.最后，将这三个文件提交到新形成的路径，然后程序运行。

MapReduce的框架结构与核心运行机制

结构

一个完整的mapreduce程序在分布式运行时有3个实例进程:

1.MRAppMaster:负责整个程序的过程调度以及状态协调

2.MapTask:负责map阶段的整个数据处理流程

3。ReduceTask:负责reduce阶段的整个数据处理流程

MR运行流程示意图

流程解析:

1.一个MapReduce启动以后，首先启动的是MRAppMaster，MRAppMaster根据本次的job信息，计算出需要的MapTask数量，然后向集群申请机器启动相应的数量的MapTask进程。

2.maptask启动之后根据给定的数据范围进行数据处理，主要的流程是

A.使用程序中指定的inputformat来获取RecordRead进行读取数据，获取KV键值对数据

B.将输入的键值对输入到map()方法中，进行业务逻辑处理然后输出到缓存

C.将缓存中的KV对按照K排序后溢出写入磁盘

3.MRAppMaster监控所有的Maptask任务进程完成之后，根据程序中设置的ReduceTask数量，告知ReduceTask需要处理的数据范围。

4.ReduceTask启动进程之后，根据MRAppMaster指定的MapTask计算出来的数据的位置，从若干台MapTask运行所在机器上获取若干个MapTask输出结果文件，并在本地进行重新归并排序，然后根据相同的key的KV一个组，调用程序中的reduce()进行业务逻辑处理，然后根据用户自定的OutPutFormat将结果写出到本地。

MapTask的并行度决定机制

MapTask的并行度决定机制

一个job的maptask并行度是由客户端提交任务的时候决定的，对于job的并行度处理逻辑是：

针对于一个job的数据进行逻辑切片，这里是根据特定的大小进行逻辑划分为多个split，对于每一个split进行maptask实例化

对于切片，需要注意的是根据FileInputFormat的getSplits()方法来实现。这里的过程如下：

FileInputFormat切片机制

切片定义在FileInputFormat类中的getSplit()方法

FileInputFormat中的默认机制：

a)简单地按照文件的内容长度进行切片

b)切片大小，默认等于block大小

c)切片时不考虑数据集整体，而是逐个针对每一个文件单独切片

比如待处理数据有两个文件：

file1.txt320M

file2.txt10M

经过FileInputFormat的切片机制运算后，形成的切片信息如下

file1.txt.split1--0~128

file1.txt.split2--128~256

file1.txt.split3--256~320

file2.txt.split1--0~10M

FileInputFormat中的参数配置

通过分析源码，在FileInputFormat中，计算切片大小的逻辑：Math.max(minSize, Math.min(maxSize, blockSize));切片主要由这几个值来运算决定

minsize：默认值：1

配置参数：mapreduce.input.fileinputformat.split.minsize

maxsize：默认值：Long.MAXValue

配置参数：mapreduce.input.fileinputformat.split.maxsize

blocksize

因此，默认情况下，切片大小=blocksize

maxsize(切片最大值)：

参数如果调得比blocksize小，则会让切片变小，而且就等于配置的这个参数的值

minsize(切片最小值)：

参数调的比blockSize大，则可以让切片变得比blocksize还大

选择并发数的影响因素：

1、运算节点的硬件配置

2、运算任务的类型：CPU密集型还是IO密集型

3、运算任务的数据量

Reduce的并行度决定机制

reducetask的并行度同样影响整个job的执行并发度和执行效率，但与maptask的并发数由切片数决定不同，Reducetask数量的决定是可以直接手动设置：

//默认值是1，手动设置为4

job.setNumReduceTasks(4);

如果数据分布不均匀，就有可能在reduce阶段产生数据倾斜

MapReduce的Combiner

combiner是MR程序中Mapper和Reducer之外的一种组件

combiner组件的父类就是Reducer

combiner和reducer的区别在于运行的位置：

Combiner是在每一个maptask所在的节点运行

Reducer是接收全局所有Mapper的输出结果;

combiner的意义就是对每一个maptask的输出进行局部汇总，以减小网络传输量

具体实现步骤：

1、自定义一个combiner继承Reducer，重写reduce方法

2、在job中设置：job.setCombinerClass(CustomCombiner.class)

combiner能够应用的前提是不能影响最终的业务逻辑

而且，combiner的输出kv应该跟reducer的输入kv类型要对应起来

需要注意的是：Combiner使用需要特别注意，以为Combiner在MR过程中可能不被调用，也可能被调用一次或者多次，Combiner不使用的原则是：使不使用都不影响业务逻辑。

MapReduce的Shuffle机制

概述

mapreduce中，mapTask处理数据之后，reduceTask如何处理mapTask得出的结果，这个过程称之为shuffle。

shuffle机制：洗牌、发牌-----核心机制：数据分区、排序和缓存

具体来说：就是将mapTask处理后的数据分发给reduceTask，在分发过程中，需要数据对key进行排序和分区。

主要的流程：

主要的流程：

1.mapTask进行业务逻辑处理后，使用OutPutCollector组件将数据按照KV写入到环形缓冲区，环形缓冲区中0.8份额为保存KV，0.2份额为KV使用快速排序。

2.使用Spiller组件根据Partioner分区后文件进行溢出，用户自定义了Partioner组件，根据用户程序自定进行分区，如果没有定义根据K的值进行Hash取模。

3.对于溢出文件，对于多个小型文件，使用归并排序合并成一个大的文件。

4.reduceTask根据分区号，去各个maptask机器上取出相应分区结果

5.reduceTask会根据分区号取出不同的maptask的处理结果，reduceTask会根据归并排序进行合并形成大文件

6.合并成大文件后，shuffle的过程也就结束了，后面进入reducetask的逻辑运算过程(从文件中取出一个一个的键值对group，调用用户自定义的reduce()方法)

WordCount代码编写

//首先要定义四个泛型的类型

//keyin:LongWritablevaluein: Text

//keyout: Textvalueout:IntWritable

public class WordCountMapper extends Mapper{

//map方法的生命周期：框架每传一行数据就被调用一次

//key :这一行的起始点在文件中的偏移量

//value:这一行的内容

@Override

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

//拿到一行数据转换为string

String line = value.toString();

//将这一行切分出各个单词

String[] words = line.split(" ");

//遍历数组，输出<单词，1>

for(String word:words){

context.write(new Text(word), new IntWritable(1));

}

}

}

//生命周期：框架每传递进来一个kv组，reduce方法被调用一次

@Override

protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {

//定义一个计数器

int count = 0;

//遍历这一组kv的所有v，累加到count中

for(IntWritable value:values){

count += value.get();

}

context.write(key, new IntWritable(count));

}

}

public class WordCountRunner {

//把业务逻辑相关的信息(哪个是mapper，哪个是reducer，要处理的数据在哪里，输出的结果放哪里……)描述成一个job对象

//把这个描述好的job提交给集群去运行

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

Configuration conf = new Configuration();

Job wcjob = Job.getInstance(conf);

//指定我这个job所在的jar包

//wcjob.setJar("/home/hadoop/wordcount.jar");

wcjob.setJarByClass(WordCountRunner.class);

wcjob.setMapperClass(WordCountMapper.class);

wcjob.setReducerClass(WordCountReducer.class);

//设置我们的业务逻辑Mapper类的输出key和value的数据类型

wcjob.setMapOutputKeyClass(Text.class);

wcjob.setMapOutputValueClass(IntWritable.class);

//设置我们的业务逻辑Reducer类的输出key和value的数据类型

wcjob.setOutputKeyClass(Text.class);

wcjob.setOutputValueClass(IntWritable.class);

//指定要处理的数据所在的位置

FileInputFormat.setInputPaths(wcjob, "hdfs://hdp-server01:9000/wordcount/data/big.txt");

//指定处理完成之后的结果所保存的位置

FileOutputFormat.setOutputPath(wcjob, new Path("hdfs://hdp-server01:9000/wordcount/output/"));

//向yarn集群提交这个job

boolean res = wcjob.waitForCompletion(true);

System.exit(res0:1);

}