MapReduce被广泛应用于日志分析、海量数据排序、在海量数据中查找特定模式等场景中。

MapReduceJob

在Hadoop中，每个MapReduce任务都被初始化为一个Job。

每个Job又可以分为两个阶段：Map阶段和Reduce阶段。这两个阶段分别用Map函数和Reduce函数来表示。

Map函数接收一个<key,value>形式的输入，然后产生另一种<key,value>的中间输出，Hadoop负责将所有具有相同中间key值的value集合到一起传递给Reduce函数；Reduce函数接收一个如<key,(list of values)>形式的输入，然后对这个value集合进行处理并输出结果，Reduce的输出也是<key,value>形式的。

InputFormat()和InputSplit

InputSplit是Hadoop中用来把输入数据传送给每个单独的Map，InputSplit存储的并非数据本身，而是起始位置、分片长度和一个记录数据所在主机的数组。生成InputSplit的方法可以通过InputFormat()来设置。当数据传送给Map时，Map会将输入分片传送到InputFormat()上，InputFormat调用getRecordReader()方法生成RecordReader，RecordReader在通过creatKey()、createValue()方法将InputSplit创建成可供Map处理的<key,value>对。即，InputFormat()方法是用来生成可供Map处理的<key,value>对的。

InputFormat

BaileyBorweinPlouffe.BbpInputFormat

ComposableInputFormat

CompositeInputFormat

DBInputFormat

DistSum.Machine.AbstractInputFormat

FileInputFormat

CombineFileInputFormat

KeyValueTextInputFormat

NLineInputFormat

SequenceFileInputFormat

TeraInputFormat

TextInputFormat

TextInputFormat是Hadoop默认的输入方式。在TextInputFormat中，每个文件（或其一部分）都会单独作为Map的输入，而这是继承自FileInputFormat的。之后，每行数据都会生成一条记录，每条记录则表示成<key,value>形式。

key值是每个数据记录在数据分片中的字节偏移量，数据类型是LongWritable；

value值是每行的内容，数据类型是Text。

如：

file1:

0 hello world bye world

file2:

0 hello hadoop bye hadoop

两个文件都会被单独输入到一个Map中，因此它们的值都是0。

OutputFormat()

默认的输出格式是TextOutputFormat，每条记录以一行的形式存入文本文件，键和值是任意形式的，程序内部调用toString()方法将键和值转换为String类型再输出。

Map函数和Reduce函数：

public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{protected void map(LongWritable key, Text value,Context context)throws IOException, InterruptedException {String[] words = StringUtils.split(value.toString(), ' ');for(String w :words){context.write(new Text(w), new IntWritable(1));}}
}

public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{protected void reduce(Text key, Iterable<IntWritable> values,Context context)throws IOException, InterruptedException {int sum =0;for(IntWritable i: values){sum=sum+i.get();}arg2.write(key, new IntWritable(sum));}
}

public class RunJob {public static void main(String[] args) {Configuration config =new Configuration();
//      config.set("fs.defaultFS", "hdfs://node1:8020");
//      config.set("yarn.resourcemanager.hostname", "node1");config.set("mapred.jar", "C:\\Users\\Administrator\\Desktop\\wc.jar");try {FileSystem fs =FileSystem.get(config);Job job =Job.getInstance(config);job.setJarByClass(RunJob.class);job.setJobName("wc");job.setMapperClass(WordCountMapper.class);job.setReducerClass(WordCountReducer.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);FileInputFormat.addInputPath(job, new Path("/usr/input/"));Path outpath =new Path("/usr/output/wc");if(fs.exists(outpath)){fs.delete(outpath, true);}FileOutputFormat.setOutputPath(job, outpath);boolean f= job.waitForCompletion(true);if(f){System.out.println("job completion");}} catch (Exception e) {e.printStackTrace();}}
}

注意两种情况：

1、Reduce Task的数量可以由程序指定，当存在多个Reduce Task时，每个Reduce会搜集一个或多个key值。当存在多个Reduce Task时，每个Reduce Task都会生成一个输出文件；

2、没有Reduce任务的时候，系统会直接将Map的输出结果作为最终结果，有多少个Map就有多少个输出。

MapReduce任务的优化

如何完成这个任务，怎么能让程序运行的更快。

MapReduce计算模型的优化主要集中在两个方面：计算性能方面；IO操作方面。

1、任务调度；

计算方面：优先将任务分配给空闲机器；

IO方面：尽量将Map任务分配给InputSplit所在的机器。

2、数据预处理与InputSplit的大小

MapReduce擅长处理少量的大数据，在处理大量的小数据时性能会很逊色。

因此在提交MapReduce任务前可以先对数据进行一次预处理，将数据合并以提高MapReduce任务的执行效率。

另一方面是参考Map任务的运行时间，当一个Map任务只需要运行几秒就可以结束时，就需要考虑是否应该给它分配更多的数据。通常而言，一个Map任务的运行时间在一分钟左右比较合适。

在FileInputFormat中（除了CombineFileInputFormat），Hadoop会在处理每个Block后将其作为一个InputSplit，因此合理地设置block块大小是很重要的调节方式。

3、Map和Reduce任务的数量

Map/Reduce任务槽：集群能够同时运行的Map/Reduce任务的最大数量。

如100台机器，每台最多同时运行10个Map和5个Reduce，则Map任务槽为1000，Reduce任务槽为500。

设置Map任务的数量主要参考的是Map的运行时间，设置Reduce任务的数量主要参考的是Reduce槽的数量。

Reduce任务槽的0.95倍，如果一个Reduce任务失败，可以很快找到一个空闲的机器重新执行；

Reduce任务槽的1.75倍，执行快的机器可以获得更多的Reduce任务，因此可以使负载更加均衡，以提高任务的处理速度。

4、Combine函数

用于本地合并数据，以减少网络IO操作的消耗。

合理的设计combine函数会有效减少网络传输的数据量，提高MapReduce的效率。

job.setCombinerClass(combine.class);

在WordCount中，可以指定Reduce类为combine函数：

job.setCombinerClass(Reduce.class);

5、压缩

可以选择对Map的输出和最终的输出结果进行不同压缩方式的压缩。

在一些情况下，Map的中间输出可能会很大，对其进行压缩可以有效地减少网络上的数据传输量。

6、自定义comparator

自定义Hadoop数据类型时，推荐自定义comparator来实现数据的比较，这样可以省去数据序列化和反序列化的时间，提高程序的运行效率。

3 MapReduce计算模型相关推荐

从 WordCount 到 MapReduce 计算模型
概述虽然现在都在说大内存时代,不过内存的发展怎么也跟不上数据的步伐吧.所以,我们就要想办法减小数据量.这里说的减小可不是真的减小数据量,而是让数据分散开来.分开存储.分开计算.这就是 MapRedu ...
大数据Hadoop（十六）：MapReduce计算模型介绍
文章目录 MapReduce计算模型介绍理解MapReduce思想 Hadoop MapReduce设计构思
大数据图数据库之离线挖掘计算模型
/* 版权声明:可以任意转载,转载时请务必标明文章原始出处和作者信息 .*/ author: 张俊林节选自<大数据日知录:架构与算法>十四章,书籍目录在此对于离 ...
大数据计算技术-数据计算处理系统-计算模式（计算模型）
大数据计算技术-数据计算处理系统-计算模式(计算模型) 1计算模式(计算模型) 1.1批处理计算模式:数据计算处理系统例子:MapReduce计算模型 1.2流计算模式 1.2.1定义 1.2.2流计 ...
Hadoop详解（六）：MapReduce计算框架详解
1. Hadoop MapReduce简介 Hadoop MapReduce是一个使用简便的软件框架,是Google云计算模型MapReduce的Java开源实现,基于它写出来的应用程序能够运行在由上 ...
Storm 02_Storm 计算模型
一.Storm 计算模型 Topology – DAG有向无环图的实现对于Storm实时计算逻辑的封装即,由一系列通过数据流相互关联的Spout.Bolt所组成的拓扑结构生命周期:此拓扑只要启动 ...
大数据-MapReduce计算框架
导语 MapReduce作为Hadoop核心编程模型,在Hadoop中,数据处理的核心就是MapReduce程序设计模型.下面就来分享一下MapReduce都有那些值得我们注意的事情. 文章目录 ...
BigData之Hadoop：Hadoop框架(分布式系统基础架构)的简介(两大核心【HDFS存储和MapReduce计算】)、深入理解、下载、案例应用之详细攻略
BigData之Hadoop:Hadoop框架(分布式系统基础架构)的简介(两大核心[HDFS存储和MapReduce计算]).深入理解.下载.案例应用之详细攻略目录 Hadoop的简介(分布式系统 ...
Twister: 迭代MapReduce计算框架
摘要:MapReduce编程模型已经简化了许多数据并行应用的实现.编程模型的简化和MapReduce实现提供的服务质量在分布式计算社区上吸引了很多的热情.把MapReduce应用到多种科学应用的这么多 ...

3 MapReduce计算模型