Hadoop中MapReduce的使用

1 概述

MapReduce是一个分布式运算程序的编程框架，是用户开发“基于Hadoop的数据分析应用”的核心。其功能是将用户编写的业务逻辑代码和自带默认组件整合成一个完整的分布式运算程序，并发运行在一个Hadoop集群上。

它的主要优点有：

1. 易于编程：它提供了简单易用的框架接口供人调用，开发人员只需关注业务逻辑的实现，而不必关心底层任务分发和收集的MapReduce实现的相关细节
2. 良好扩展性：计算资源够的时候，可以动态增加机器来扩展计算能力。
3. 高容错性：任何一台机器挂了，它可以把上面的计算任务转移到其他节点上运行
4. 海量计算：可以实现上千台服务器集群并发工作，提供海量数据处理能力。

但也存在缺点：

1. 不擅长实时计算：MapReduce无法像MySQL一样，在毫秒或者秒级内返回结果。
2. 不擅长流式计算：流式计算的输入数据是动态的，而MapReduce的输入数据集是静态的。无法像SparkStreaming和Flink一样处理流式数据。
3. 不擅长DAG（有向无环图）计算：DAG计算是指后一个应用程序的输入需要前一个的输出。由于MapReduce的输出结果会写入磁盘，无法像Spark一样将中间结果放入内存，造成大量的磁盘IO，导致性能非常的低下。

核心思想

如下所示，以字符统计为例说明MapReduce思想

1. 首先假设输入数据分别为200M和100M的文档，一般处理的数据块最大为128M，所以第一个文件被切分为128M+72M两份。
2. 在Map阶段将切分好的数据块分别交给不同的数据节点，每个节点分别负责统计不同的数据块。统计时将单词以a到p开头的放到分区1，以q到z开头的放到分区2.
3. 在Reduce阶段，依次遍历各个节点，分别收集统计各个节点分区1和分区2中单词的数量
4. 最后得到输出结果

MapReduce编程模型只能包含一个Map阶段和一个Reduce阶段，如果用户的业务逻辑非常复杂，那就只能多个MapReduce程序，串行运行。

一个完整的MapReduce程序在分布式运行时有三类实例进程：
（1）MrAppMaster：负责整个程序的过程调度及状态协调。
（2）MapTask：负责Map阶段的整个数据处理流程。
（3）ReduceTask：负责Reduce阶段的整个数据处理流程。

2 WordCount案例

编写程序统计如下文本中每个单词出现的次数，

Hadoop
Hadoop File System
Map Reduce
Spark Flink Storm
Spark Stream

按照上面MapReduce的思想，一个MapReduce程序主要包含三个类：Mapper、Reducer和Driver。在Mapper阶段将文件划分到不同节点分别统计a到p、q到z开头的单词，然后在Reduce阶段对各个节点的统计结果进行汇总。

2.1 Mapper

用户在自定义的Mapper类中完成任务中的切分 *** 作，它需要继承父类Mapper，并实现map()方法，在其中实现业务逻辑 *** 作。Mapper的输入输出都是以键值对的形式，可以通过泛型的方式对四个参数类型进行设置，并且对每个键值对都会调用一次map()方法。Mapper在遍历之前会执行一次setup()方法，可以定义一些初始化 *** 作；在遍历完成之后执行一次cleanup()方法，可以在其中定义一些资源关闭等 *** 作。

如下所示的WordCountMapper类中，输入为，例如第二行文本<2, Hadoop File System>。在map()方法中获取到字符串后首先按空格切分为单词，然后以输出到context中，其中1代表hadoop出现了一次。

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 line = value.toString();
		
		// 2 按照空格进行切割
		String[] words = line.split(" ");
		
		// 3 输出键值对到context
		for (String word : words) {
			k.set(word);
			context.write(k, v);
		}
	}
}

2.2 Reducer

Reducer类的输入输出也是以键值对的方式，也需要在继承的时候对泛型类型进行设置。主要在reduce()方法中实现主要的功能逻辑，Mapper输出的多个键值对会在进入Reducer后进行合并，相同键的值会被放在一起，例如就是把以hadoop为键的所有值放在了一个迭代器里，在reduce()方法中遍历并累加就可以得出hadoop单词出现的次数。

需要注意的是Hadoop中的迭代器使用了对象重用，迭代时value始终指向一个内存地址，改变的是引用指向的内存地址中的数据。因此当涉及到对象的复制 *** 作时不能直接复制value，而需要通过BeanUtils.copyProperties()将其中的内容拷贝出来。

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

2.3 Driver

Driver类主要完成和Mapper与Reducer类的关联工作，并在其中对输入输出进行设置。并且通过FileInputFormat对输入输出的文件进行设置，在运行时以参数的形式传入文件名。注意输出文件的路径不能是已存在的文件夹

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类
		job.setMapperClass(WordCountMapper.class);
		job.setReducerClass(WordCountReducer.class);

		// 4 设置Mapper输出的键值对类型
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);

		// 5 设置最终输出键值对类型
		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
		boolean result = job.waitForCompletion(true);
		System.exit(result ? 0 : 1);
	}
}

将项目打包为WordCount.jar并上传到集群中，执行WordCount.jar，这里需要精确到WordCountDriver 类的位置，并且将输入和输出文件夹作为参数传入，这里使用的是hdfs上的位置

hadoop jar WordCount.jar mapreduce.WordCountDriver /document /output/WordCount

3 序列化对象

为了便于在不同服务器之间进行对象的传输，需要通过序列化的方式将内存中的对象转换成字节序列，在接收到数据之后再通过反序列化转换成内存中的对象。

Java提供了一趟默认的序列化方式Serializable，一个对象被序列化后，会附带很多额外的信息（各种校验信息，Header，继承体系等），不便于在网络中高效传输。所以，Hadoop自己开发了一套序列化机制（Writable）。

MapReduce使用数据的类型是Hadoop自身封装的序列化类型，如下所示，除了String外，其他类型都是Java基本类型后+Writable

Java类型	Hadoop Writable类型
Boolean	BooleanWritable
Byte	ByteWritable
Int	IntWritable
Float	FloatWritable
Long	LongWritable
Double	DoubleWritable
String	Text
Map	MapWritable
Array	ArrayWritable
Null	NullWritable

除了以上基本数据类型之外，在开发中有可能自定义一些数据对象，我们需要实现Writable接口以便于对象的序列化和反序列化。

如下所示，以统计手机流量为例，对流量对象实现序列化。输入数据如下：

id	手机号码			网络ip				上行流量  下行流量     网络状态码
1	13736230513	192.196.100.1	www.atguigu.com	2481	24681	200
2	13846544121	192.196.100.2			264	0	200
3 	13956435636	192.196.100.3			132	1512	200
4 	13966251146	192.168.100.1			240	0	404
5 	18271575951	192.168.100.2	www.atguigu.com	1527	2106	200
6 	84188413	192.168.100.3	www.atguigu.com	4116	1432	200
7 	13590439668	192.168.100.4			1116	954	200
8 	15910133277	192.168.100.5	www.hao123.com	3156	2936	200
9 	13568436656	192.168.100.18	www.alibaba.com	2481	24681	200
10 	13568436656	192.168.100.19			1116	954	200

通过对其中相同手机号码流量的统计，输入如下格式的数据

13560436666 		1116		      954 			2070
手机号码		    上行流量        下行流量		总流量

在Map阶段获取一行输入数据之后，首先按照/t对数据进行切分，然后获得其中的手机号码、上行流量和下行流量。之后以手机号为key，流量信息为value作为Map阶段的输出。最后在Reduce阶段对流量信息进行累加即可得到结果。

但是流量信息并不是单纯的一个整型或字符串，而是一个复杂的Java对象，如下所示通过FlowBean来储存流量信息。

作为一个Java Bean，FlowBean首先要提供无参构造函数和属性字段的getter和setter方法；作为Hadoop的序列化对象，首先要继承Writable接口，并实现其序列化和反序列化方法write()和readFileds()，注意数据的写入和接收时读取的顺序要保持一致；最后为了打印输出结果，需要重写toString()方法。

如果Key值也需要自定义的JavaBean，则还需要实现Comparable接口并重写compareTo()方法，因为MapReduce框中的Shuffle过程要求对key必须能排序。

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

之后在Mapper类中使用FlowBean作为输出value的类型，并在其中读取手机号和流量数据，封装到到输出的键值对中

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

在Reducer类中以作为输入，读取其中的流量信息完成累加

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

最后在Driver中关联Mapper和Reducer，设置输入输出后提交任务

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