hadoop之MapReduce简介

一、MapReduce概述

1、MapReduce定义

MapReduce是一个分布式运算程序的编程框架，是用户开发“基于Hadoop的数据分析应用”的核心框架。MapReduce核心功能是将用户编写的业务逻辑代码和自带默认组件整合成一个完整的分布式运算程序，并发运行在一个Hadoop集群上。
简单说MapReduce是一个框架，一个分布式计算框架，只需用户将业务逻辑放到框架中，就会和框架组成一个分布式运算程序，在Hadoop集群上实行分布式计算。
MapReduce的核心思想就是将大数据的任务，分解成多个小数据的任务，交由Map分布式处理，最后再由Reduce合并结果。

2、MapReduce的优缺点

优点：
（1）MapReduce易于编程，简单的实现一些接口，即可完成一个分布式程序
（2）良好的扩展性，当计算资源不足时，可以通过简单的增加廉价的机器来扩展计算能力
（3）高容错性，当一个任务计算失败时，可以将失败的计算任务转移到另外一个节点运行
（4）适合TB，PB以上海量数据的离线处理
缺点：
（1）不擅长实时计算，现在的实时计算框架，由Flink完成
（2）不擅长流式计算，流式计算的输入数据是动态的，而MapReduce的输入数据集是静态的，目前流式计算由Flink或者spark完成
（3）不擅长DAG（有向无环图）计算，即一个计算的结果，作为下一个计算的输入，现在的流式计算框架由Spark完成，因为Spark的计算结果存储在内存，而MapReduce的计算结果存在磁盘中，每次输出的结果都存在磁盘，会导致频繁的IO，使MapReduce的性能比较低。
Spark 和 Flink 都是分布式计算框架，但他们都是基于内存的，所以计算的速度要优于MapReduce。

3、MapReduce的进程

MapReduce的实例进程一般为三部分：
（1）MRAppMaster：负责整个MR程序的过程调度以及和 ResourceManager 的交互，一个MapReduce只开启1个。
（2）MapTask：负责Map阶段的过程调度以及具体实施，一般为1个到多个，根据切片数量来决定开启数量。
（3）ReduceTask：负责Reduce阶段的数据合并处理，一般为0个到多个，当数据在Map阶段就能合并时，Reduce可以不用开启。
其中 ResourceManager 就是Yarn的管理者，就是资源管理器的管理者，简称为RM。

MapReduce的代码处理过程分为三个阶段：
（1）Mapper 阶段
Map阶段将大的处理任务分为小任务，然后交由各个节点独立运行，互不干扰。
（2）Reduce 阶段
Reduce阶段将Map阶段的运行结果做汇总。
（3）Driver 阶段
Driver相当于Yarn集群的客户端，用于提交整个MapReduce程序到Yarn集群运行，提交的是封装了MapReduce程序相关运行参数的Job对象。因为所有的MapReduce最终都是交由节点来运行的，而具体分配到哪个节点，就由Yarn来做资源分配。

4、MapReduce的编程规范

Map阶段：
（1）用户自定义的Mapper需要继承Mapper的父类，extends Mapper<k,v,k,v>
（2）Mapper的输入数据是Key-Value对（健值对）的形式（KV的泛型类型需要根据业务逻辑来确定）
（3）Mapper中的业务逻辑是写在map()方法中的，重写父类的map方法来实现
（4）Mapper的输出数据也是Key-Value对（键值对）的形式（KV的泛型类型需要根据业务逻辑来确定）
（5）输出的健值对，通过context.write写入到上下文中
（6）针对每一对<key,value>都会调用一次map()方法（MapTask进程）
Reduce阶段：
（1）用户自定义的Reduce需要继承Reduce的父类
（2）Reduce的输入数据是KV对的形式，同时也是mapper阶段的输出数据，这里的健值对必须跟Map阶段的键值对类型一致，Mapper的输出，就是这里的输入。
（3）Reduce中的业务逻辑是写在reduce()方法中的，重写父类的reduce方法来实现
（4）输出的健值对，通过context.write写入到上下文中
（5）ReduceTask进程对每一组相同的key调用一次reduce()方法
Driver阶段：
（1）获取配置信息，获取Job对象实例
（2）关联本Driver的jar包
（3）关联mapper和reducer的jar包
（4）设置mapper的输出健值参数
（5）输出最终输出的健值参数，不一定是Reduce的，有些没有Reduce或其他
（6）设置输入和输出路径
（7）提交job到yarn运行
（8）根据其他需要设置，比如设置分区等

5、hadoop的数据类型

其中只有java中的String类型，对应hadoop的类型写法不同，其他的都是原类型+Writable的写法，同时也要注意两种类型的切换。

6、wordCount的案例演示

wordCount就是计算每个单词出现的次数，假设有文件 words.txt 内容如下：

hello hadoop
hello map
hello reduce
hello map reduce
hadoop hadoop

程序思路分析：

mapper阶段：
（1）每次读取1行，hadoop中读取的文本为text，将类型转为string
（2）将每一行String根据空格进行拆分，将每个单词存到String类型的数组中
（3）取出每个单词，合并成<单词，1>的键值对，1代表该单词出现的次数
因为这里的输入是每一行，所以当数据量很大时，可以按行将任务划分为小任务，符合分布式思想，且怎么划分都不会影响后续的计算结果。
reducer阶段：
（1）读取所有<单词，1>的键值对的值
（2）根据<单词，1>的键值对，对每个相同的单词，对后面的数字1进行累加即可计算该单词的次数
driver阶段：
（1）获取配置信息，获取JOB对象
（2）关联本Driver的jar包
（3）关联mapper和reducer的jar包
（4）指定mapper输出类型的kv类型
（5）指定最终输出的数据的kv类型
（6）指定JOB输入和输出文件的路径
（7）提交作业

程序实现：

（1）创建maven工程，MapReduceDemo
（2）在pom.xml文件中添加如下依赖

<dependencies>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>3.1.3</version>
    </dependency>
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>1.7.30</version>
    </dependency>
</dependencies>

（3）在项目的src/main/resources目录下，新建一个文件，命名为“log4j.properties”，在文件中填入。

log4j.rootLogger=INFO, stdout  
log4j.appender.stdout=org.apache.log4j.ConsoleAppender  
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout  
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n  
log4j.appender.logfile=org.apache.log4j.FileAppender  
log4j.appender.logfile.File=target/spring.log  
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout  
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

（4）创建包名：com.mapreduce.wordcount
（5）编写程序
编写Mapper类

package com.mapreduce.wordcount;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
	Text k = new Text();
	IntWritable v = new IntWritable(1);
	@Override
	protected void map(LongWritable key, Text value, Context context)	throws IOException, InterruptedException {
		// 1 获取一行并将其转成String类型来处理
		String line = value.toString();
		// 2 将String类型按照空格切割后存进String数组
		String[] words = line.split(" ");
		// 3 依次取出单词，将每个单词和次数包装成键值对，写入context上下文中供后续调用
		for (String word : words) {
			// 先将String类型，转为text，再包装成健值对
			k.set(word);
			context.write(k, v);
		}
	}
}

Mapper<LongWritable, Text, Text, IntWritable>继承Mapper父类的时候，需要加上泛型，这里有两对键值对 <LongWritable, Text,> 和 <Text, IntWritable> 第一个键值对表示输入的数据，LongWritable表示输入数据的索引，就是类似于第几行数据，Text表示读入的内容，就是当读取文件时，系统会将文件的索引和该行的值赋给这两个泛型。第二个键值对表示输出的数据，Text表示输入的单词，IntWritable表示该单词的次数，一般第一个健值对可以按照系统默认，第二个键值对需要根据业务逻辑来确定。

编写Reducer类

package com.mapreduce.wordcount;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
	int sum;
	IntWritable v = new IntWritable();
	@Override
	protected void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException {
		// 1 累加求和
		sum = 0;
		for (IntWritable count : values) {
			sum += count.get();
		}
		// 2 输出
         v.set(sum);
		context.write(key,v);
	}
}

Reducer继承父类的时候，这里的泛型<Text, IntWritable, Text, IntWritable>，其中第一对泛型<Text, IntWritable>就是Mapper阶段的输出泛型，表示每个单词和出现的次数，第二个泛型就是需要输出的结果数据，这里需要输出每个单词的出现次数，就是将次数叠加之后，还是int类型，所以还是<Text, IntWritable>类型。
上面有说到，Reduce是每组会执行一次，就是相同的key是会分到同一组的，所以此处只需计算每个key的count叠加即可。

编写Driver驱动类

package com.mapreduce.wordcount;
import java.io.IOException;
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;

public class WordCountDriver {
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		// 1 获取配置信息以及获取job对象
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);
		// 2 关联本Driver程序的jar
		job.setJarByClass(WordCountDriver.class);
		// 3 关联Mapper和Reducer的jar
		job.setMapperClass(WordCountMapper.class);
		job.setReducerClass(WordCountReducer.class);
		// 4 设置Mapper输出的kv类型
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		// 5 设置最终输出kv类型，此处是reduce的kv对类型输出
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		// 6 设置输入和输出路径
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		// 7 提交job到yarn运行
		boolean result = job.waitForCompletion(true);
		System.exit(result ? 0 : 1);
	}
}

二、hadoop序列化

1、序列化的定义

序列化就是把内存中的对象，转换成字节序列（或其他数据传输协议）以便于存储到磁盘（持久化）和网络传输。
反序列化就是将收到字节序列（或其他数据传输协议）或者是磁盘的持久化数据，转换成内存中的对象。
上面在介绍MapReduce的优点时，说的是MapReduce可以在任务处理失败的时候，会将失败的任务转给其他节点执行。每个节点就是每个服务器，在A服务器执行失败的任务，我们需要将任务交给B服务器来执行。这里面有个问题就是，我们在A服务器中创建的对象是在A服务器的内存中的，那内存中的对象要怎么传递到B服务器中呢。如果这个对象的类型就是我们上面说的基本类型，IntWritable等，hadoop已经做好序列化了。但如果对象是自定义的，这时就需要序列化了，将自定义的对象类型按照定义序列化后转成字节序列，到了B服务器再将其反序列化，转成对象，这样就能实现在不同节点直接传递任务，就能做到高容错性了。

2、java和hadoop的序列化对比

Java的序列化是一个重量级序列化框架，一个对象被序列化后，会附带很多额外的信息（各种校验信息，Header，继承体系等），不便于在网络中高效传输。所以，Hadoop自己开发了一套序列化机制（Writable），以便更方便快捷的实现序列和反序列化。
hadoop自己的序列化相对java的序列化拥有以下优点：
（1）紧凑 ：高效使用存储空间。
（2）快速：读写数据的额外开销小。
（3）互操作：支持多语言的交互。

3、序列化的操作步骤

序列化的步骤：
（1）序列化对象必须实现Writable接口，implements Writable
（2）反序列化时，需要调用空参构造函数，所以序列化对象必须要有空参构造方法
（3）重写序列化方法write()
（4）重写反序列化方法readFields()
（5）注意反序列化的顺序和序列化的顺序必须完全一致
（6）要想把结果显示在文件中，需要重写toString()，可用"\t"分开，方便后续用。
（7）如果需要将自定义的序列化对象放在key中传输，则还需要实现Comparable接口，因为MapReduce框架中的Shuffle过程要求对key必须能排序

序列化简单示例：
（1）编写序列化对象

package com.mapreduce.writable;
import org.apache.hadoop.io.Writable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

//1 继承Writable接口
public class FlowBean implements Writable {
    private long upFlow; //上行流量
    private long downFlow; //下行流量
    private long sumFlow; //总流量
    //2 提供无参构造
    public FlowBean() {
    }
    //3 提供三个参数的getter和setter方法
    public long getUpFlow() {
        return upFlow;
    }
    public void setUpFlow(long upFlow) {
        this.upFlow = upFlow;
    }
    public long getDownFlow() {
        return downFlow;
    }
    public void setDownFlow(long downFlow) {
        this.downFlow = downFlow;
    }
    public long getSumFlow() {
        return sumFlow;
    }
    public void setSumFlow(long sumFlow) {
        this.sumFlow = sumFlow;
    }
    public void setSumFlow() {
        this.sumFlow = this.upFlow + this.downFlow;
    }
    //4 实现序列化和反序列化方法,注意顺序一定要保持一致
    @Override
    public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeLong(upFlow);
        dataOutput.writeLong(downFlow);
        dataOutput.writeLong(sumFlow);
    }
    @Override
    public void readFields(DataInput dataInput) throws IOException {
        this.upFlow = dataInput.readLong();
        this.downFlow = dataInput.readLong();
        this.sumFlow = dataInput.readLong();
    }
    //5 重写ToString
    @Override
    public String toString() {
        return upFlow + "\t" + downFlow + "\t" + sumFlow;
    }
    // 6 如果作为Key传输，则还需要实现compareTo方法
    //@Override
	//public int compareTo(FlowBean o) {
		// 倒序排列，从大到小
		//return this.sumFlow > o.getSumFlow() ? -1 : 1;
	//}
}

（2）编写Mapper类

package com.mapreduce.writable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;

public class FlowMapper extends Mapper<LongWritable, Text, Text, FlowBean> {
    private Text outK = new Text();
    private FlowBean outV = new FlowBean();
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //1 获取一行数据,转成字符串
        String line = value.toString();
        //2 切割数据
        String[] split = line.split("\t");
        //3 抓取我们需要的数据:手机号,上行流量,下行流量
        String phone = split[1];
        String up = split[split.length - 3];
        String down = split[split.length - 2];
        //4 封装outK outV
        outK.set(phone);
        outV.setUpFlow(Long.parseLong(up));
        outV.setDownFlow(Long.parseLong(down));
        outV.setSumFlow();
        //5 写出outK outV
        context.write(outK, outV);
    }
}

（3）编写Reducer类

package com.mapreduce.writable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;

public class FlowReducer extends Reducer<Text, FlowBean, Text, FlowBean> {
    private FlowBean outV = new FlowBean();
    @Override
    protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {
        long totalUp = 0;
        long totalDown = 0;
        //1 遍历values,将其中的上行流量,下行流量分别累加
        for (FlowBean flowBean : values) {
            totalUp += flowBean.getUpFlow();
            totalDown += flowBean.getDownFlow();
        }
        //2 封装outKV
        outV.setUpFlow(totalUp);
        outV.setDownFlow(totalDown);
        outV.setSumFlow();
        //3 写出outK outV
        context.write(key,outV);
    }
}

（4）编写Driver驱动类

package com.mapreduce.writable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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;

public class FlowDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        //1 获取job对象
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        //2 关联本Driver类
        job.setJarByClass(FlowDriver.class);
        //3 关联Mapper和Reducer
        job.setMapperClass(FlowMapper.class);
        job.setReducerClass(FlowReducer.class);
		//4 设置Map端输出KV类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(FlowBean.class);
		//5 设置程序最终输出的KV类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);
		//6 设置程序的输入输出路径
        FileInputFormat.setInputPaths(job, new Path("D:\\inputflow"));
        FileOutputFormat.setOutputPath(job, new Path("D:\\flowoutput"));
		//7 提交Job
        boolean b = job.waitForCompletion(true);
        System.exit(b ? 0 : 1);
    }
}

三、MapReduce框架原理

1、框架原理

上面说MapReduce可以分为三个阶段，那是代码的编写和处理划分，其实MapReduce的框架可以划分为五大部分，分别是inputFormat，mapper，shuffle，reducer和outputFormat。
其中inputFormat负责文件的读取方式和切块的方式；mapper就是将大任务分成小任务让节点执行；shuffle就是对mapper之后reduce之前的数据进行分区和排序，key在MapReduce框架中必须排序；reduce就是将mapper阶段的数据进行合并；outputFormat就是控制输出文件的格式，是以压缩的方式存储还是以特定的方式输出来作为下一个mapper的输入等。这就是MapReduce的整个处理的流程。

2、inputFormat

就是数据输入的格式，主要涉及两种，1是数据的读取方式，2是数据的切片方式；

数据读取

数据读取的方式主要针对不同的文件格式（如日志文件，二进制文件，数据库表等）和不同的读取方式（如单行读取，多行读取，还是多个文件合并读取等），这里只介绍部分常见的接口，FileInputFormat常见的接口包括TextInputFormat（文本输入）、KeyValueTextInputFormat（健值输入）、NLineInputFormat（多行输入）、CombineTextInputFormat（合并输入）和自定义InputFormat（自定义输入）等。
（1）TextInputFormat 是默认的FileInputFormat实现类，也是hadoop默认的数据读取方式，按行读取，一般效率较低。健值对<key,value>中的Key是存储该行在整个文件中的偏移量，为LongWritable类型，值Value是这行的内容，为Text类型
（2）NLineInputFormat 多行读取文件，效率比textInputFormat高，一次读取多行数据
（3）CombineTextInputFormat 多文件合并读取，这个需要设置最大切片数值，默认是4M，由CombineTextInputFormat.setMaxInputSplitSize(job, 4194304) 来设置大小。
例如有文件：

规则如下：
1.按照4M比较，大于4M，则对半切分，切分后为

2.按照顺序合并，A跟B1合在一起，B2跟C1合在一起，C2跟D合在一起，最后就是3个文件切片。

当然4M（4x1024x1024）这个值可以调整，当调整成20M时，4个文件就会合成1个文件处理，切片数就是1个，可以提高资源的利用，因为容器资源的申请是1G，只需开辟一个容器即可，不用按照4个文件开启4个容器。（容器的概念是在yarn中的资源调配时会介绍）

数据的切片

数据块与数据切片：
数据块：Block是HDFS物理上把数据分成一块一块来存储，数据块是HDFS存储数据的单位。
数据切片：数据切片只是在逻辑上对输入进行分片，并不会在磁盘上将其切分成片进行存储。数据切片是MapReduce程序计算输入数据的单位，一个切片会对应启动一个MapTask。
切片规则：
系统默认切片时不考虑数据集整体，而是针对每一个文件单独切片，如果文件较小，则文件单独为1片；如果文件较大，则按块大小来切分；这样当小文件很多时，系统默认多少个小文件就切多少切片，这样就会起多少个MapTask，当我们资源不足时，其实这种效率是很低下的。因为当我们8个2K的小文件时，系统需要起8个MapTask，每个的内存默认是1G，就会占用8G的内存，其实只需起1个MapTask就能处理8*2K的文件内容了。
（1）一个Job的Map阶段并行度由客户端在提交Job时的切片数决定
（2）每一个切片分配一个MapTask并行实例处理
（3）默认情况下，切片大小=BlockSize大小，HDFS默认是128M或256M，本地模式默认是32M，因为一个块是存储在一个节点上面，所以当本地节点处理本地数据的时候效率最高，如果同时还要处理其他节点过来的数据，就会效率变低，所以一般按块大小处理
（4）切片时不考虑数据集整体大小，而是逐个针对每一个文件单独切片，即文件能切就切，不能切就按照一个切片处理
（5）切片时考虑切完后大小是否是小于切片后的1.1倍，如果小于，就不切片，按照一个处理。如文件大小33M，块大小32M，33小于32的1.1倍，此时不会切成2片，而是按照1片处理。
（6）切完后的切片信息保留在客户端（此外还有xml配置文件和jar包），在申请执行mapper时，会上传给MRAppMater
（7）调整切片大小，需调小，可以设置Maxsize值，需调大，可以设置MinSize值
（8）代码中整个切片核心过程都在getSplit()里面完成

3、mapper

map阶段就是将大数据任务分成小数据任务，并行分布式处理。

4、shuffle

shuffle其实就是在mapper之后，reduce之前对数据在内存中的一些处理操作，一般分为分区和排序。
大概的流程如下：
（1）MapTask收集mapper输出的KV对，放到内存缓冲区
（2）如果内存缓冲区不断溢出数据，就会将溢出数据调用partitioner进行分区和对key排序，如果内存不溢出，则map结束时一次性将所有数据调用partitioner进行分区和对key排序
（3）如果数据很大，不断溢出时，会将多个溢出文件合并成大文件
（4）ReduceTask根据自己的分区号，主动去MapTask机器上面取得相应的结果分区数据
（5）ReduceTask会抓取同一个分区的来自不同MapTask的结果文件，并将这些文件再次进行合并和排序
（6）合并成大文件，或者一次性将所有数据都调出分区和排序后，shuffle过程也即结束
Shuffle的缓冲区大小会影响MapReduce程序的执行效率，原则上缓冲区越大，减少了IO次数，效率越高。但对内存的占用也越大，当数据量较大时，可以作为其中一个调优的策略。
参数调整为：mapreduce.task.io.sort.mb 默认是100M的大小

分区 partitioner

分区就是按照不同的条件将数据输出到不同的分区中，是为了将同种类型的数据按照统一的规格分配到对应的reducer中进行处理。
分区的数量由numReduceTasks决定，代码中的表现如下：

return key.hashCode() % numReduceTasks;

所以控制numReduceTasks的值就能控制分区的数量，并且分区是从0开始的，只能顺序递增，因为上面的代码是取模。当不主动设置分区时，系统默认都是1个分区，并且分区里面也是会排序。

自定义设置分区的步骤如下：
（1）自定义类如abc.class继承Partitioner类，重写getPartition()方法

public class abc extends Partitioner<Text, FlowBean> {
	@override
	public int getPartition(Text key, FlowBean value, int numPartitions) {
		// 控制分区代码逻辑
		...
		return partition;
	}
}

（2）在job驱动中设置自定义的Partitioner

job.setPartitonerClass(abc.class);

（3）自定义partitioner后，要根据自定义的Partitioner的逻辑设置相应的reduceTask数量

job.setNumReduceTasks(5);

注意：但设置的ReduceTasks数量跟getPartition数量不一致时，就是相当于设置的分区数量跟开启的reduce数量不一致时：
（1）如果setNumReduceTasks(5) > getPartition分区的数量，则会多出几个空的输出文件part-r-000xx
（2）如果setNumReduceTasks(5) < getPartition分区的数量，则会有多出的文件无处安放，系统会抛异常信息
（3）如果setNumReduceTasks(1)数量为1，则不管getPartition分区的数量设置为多少个，最终文件都只会产生一个结果文件，part-r-0000
（4）分区号必须从0开始，并逐一累加，因为上面分区号的取值由return key.hashCode() % numReduceTasks的取模得出，是按照0开始顺序取出的，不按照从0开始，分区号对不上，势必会报错。

排序

MapTask和ReduceTask均会对数据按照Key进行排序，该操作属于Hadoop的默认行为，任何应用程序中的数据均会被排序，而不管逻辑上是否需要，排序也是MapReduce框架中重要的操作之一。
对于shuffle，也就是map之后的数据，会先将结果暂时存放在环形缓冲区，达到一定阈值或者Map结束之后，会对缓冲区中的数据按照字典顺序按照快速排序的方式进行排序。排完序后直接写到磁盘上，如果是达到阈值后溢写的数据，则当整个Map结束后，会对缓冲区和磁盘上的数据再进行一次归并排序，以达到整体数据有序的状态。
（1）mapper处理之后的数据，分区内进行排序，分区合并的时候也会进行排序，排序按照字典顺序，用的算法是快速排序算法。
（2）快速排序是按照key的索引来排序的，而不是key的内容
（3）环形缓冲区默认是100M，分两边存储，一遍存储索引，一边存储数据，阈值默认是80%
（4）排序的规则是按照字典的规则来排序的
（5）如果不超出缓冲区的80%，直接在内存快速排序，排完后直接写出到磁盘
（6）如果超出缓冲区的80%，则将内存的数据快速排序后，写出到磁盘形成文件，每次超过阈值都会在磁盘写出文件，等磁盘达到一定阈值时，会对磁盘的文件做归并排序形成更大的文件。或者MapTask结束时，会将缓冲区的数据和磁盘的文件一起做归并排序，形成更大的文件。

排序分类：
（1）部分排序(分区排序)
MapReduce根据输入的记录的健对数据集排序，保证每个输出的文件内部有序；
（2）全排序
输出文件只有1个，且文件内部有序，实现方式是只实现一个reduceTask，实际使用该方式效率极低，一般不采用；
（3）辅助排序GroupingComparator
在Reduce端对数据进行分组；
（4）二次排序
在上面排序的里面对另外一个字段再次排序。

combiner（可选）

可选组建combine是将一些可以合并的操作在传输给reducer之前已经合并好，这样可以减少map之后传递给reducer的数据量，可以提升效率。比如健值对<abc,1>出现1万次，可以合并成<abc, 10000>，但combine只局限于一些可以合并的情况，一些求平均数之类的就不能合并策略。
（1）Combiner是MR程序中Mapper和Reducer之外的一种组件。
（2）Combiner组件的父类就是Reducer
（3）Combiner跟Reducer的区别在于运行的位置，Combiner是在每一个MapTask所在的节点运行，Reducer是接收全局所有MapTask的输出结果进行合并
（4）Combiner的意义就是对每一个MapTask的输出进行局部汇总，以减少网络传输流量
（5）Combiner能够应用的前提是不能影响最终的业务逻辑，而且Combiner输出的键值对跟Reducer输入的键值对必须一致

5、reducer

reducer细分可以分为三个阶段：
（1）copy阶段（数据复制）
ReduceTask从各个MapTask上远程拷贝一片数据，并针对某一片数据，如果其大小超过一定阈值，则写到磁盘上，否则直接放到内存中。
（2）sort阶段（排序阶段）
在远程拷贝数据的同时，ReduceTask启动了两个后台线程对内存和磁盘上的文件进行合并，以防止内存使用过多或磁盘上文件过多。按照MapReduce语义，用户编写reduce()函数输入数据是按key进行聚集的一组数据。为了将key相同的数据聚在一起，Hadoop采用了基于排序的策略。由于各个MapTask已经实现对自己的处理结果进行了局部排序，因此，ReduceTask只需对所有数据进行一次归并排序即可。
（3）reduce阶段 （处理阶段）
数据的处理和输出，reduce处理完数据后将计算结果写到HDFS上。

6、outputFormat

输出格式有MapFileOutputFormat，SequenceFileOutputFormat、TextOutputFormat等，系统默认的输出格式是TextOutputFormat，这里先不做过大介绍。

7、ReduceTask并行度

上面说的MapTask的并行度是由切片的数量来决定的，最好的情况当然是切片大小等于块大小，刚好这整个的数据都是MapTask运行的节点上。但ReduceTask的数量是可以手动设置的，虽然还要配合分区的设置。ReduceTask的设置也不是越多越好，当然数据量大时只有1个显然也发挥不了分布式的优点，所以其实ReduceTask存在一个最优或者接近最优的数量，这个无法给出准确的判断，因为这个是根据实际的业务数据量和实际的节点配置来决定的。

// 默认值是1，手动设置为4
job.setNumReduceTasks(4);

四、hadoop数据压缩

1、数据压缩概述

压缩的好处和坏处

压缩的优点：以减少磁盘IO、减少磁盘存储空间、减少网络传输流量。
压缩的缺点：增加CPU开销。

压缩原则

（1）运算密集型的Job，少用压缩
（2）IO密集型的Job，多用压缩
（3）压缩方式选择，重点考虑：压缩/解压缩速度、压缩率（压缩后存储大小）、压缩后是否可以支持切片等。

2、MR支持的压缩编码

压缩算法介绍

压缩性能的比较

由上表可知Gzip压缩：
优点：压缩率比较高；
缺点：不支持切片；压缩/解压速度一般；
Bzip2压缩：
优点：压缩率高；支持切片；
缺点：压缩/解压速度慢。
Lzo压缩：
优点：压缩/解压速度比较快；支持切片；
缺点：压缩率一般；想支持切片需要额外创建索引。
Snappy压缩：
优点：压缩和解压缩速度快；
缺点：不支持切片；压缩率一般；

3、数据压缩的使用场景

理论上来说，数据压缩可以用在MapReduce过程中的任意阶段，但一般常用的有如下的三个阶段：
一是在Map之前数据压缩；二是在Map之后，Reduce之前的数据压缩，这是为了减少Map到Reduce之间的IO或网络传输；三是Reduce之后的数据压缩。

Map之前压缩：
如果数据量小于块大小，重点考虑压缩和解压缩速度，可以使用LZO或Snappy压缩算法；
如果数据量非常大，重点考虑切片功能，考虑Bzip2和LZO，数据可以切片；
Map之后Reduce之前：
重点考虑压缩和解压缩速度可以使用LZO或Snappy，因为后续的Reduce还要用到
Reduce之后：
如果文件需要存储，则使用压缩比较高的Bzip2和Gzip，速度可以不作为考虑的优先因素。
如果文件作为下一个MR程序的输入，则就是map之前的压缩了，可以考虑数据量的大小，再考虑是否可以切片。

4、数据压缩的代码配置

压缩设置一般在Driver里面就可以设置了文章来源地址https://www.toymoban.com/news/detail-756508.html

// 开启map端输出压缩
conf.setBoolean("mapreduce.map.output.compress", true);

// 设置map端输出压缩方式
conf.setClass("mapreduce.map.output.compress.codec", BZip2Codec.class,CompressionCodec.class);

// 开启Reduce端输出压缩
FileOutputFormat.setCompressOutput(job, true);

// 设置Reduce端输出压缩的方式
 FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class);

到了这里，关于hadoop之MapReduce简介的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！