Mapreduce基本程序

1.Mapper里面的map方法

public void map(Object key,Text value,Context context) throws IOException,InterruptedException{...}

map()方法里面有三个参数，Object key,Text value就是输入的key和value,第三个参数Context context可以记录输入的key和value，此外context还会记录map运算的状态。

2.Reducer里面的reduce()方法

pubilic void reduce(Text key,Iterable<IntWritable>values,Context context) throws IOException,InterruptedException{...}

reduce()函数的输入也是一个key/value的形式，不过它的value是一个迭代器的形式Iterable<IntWritable>values，也就是说reduce中的values是一个key对应一组的值得value。

3.main()函数

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

Configuration conf = new Configuration()

String[] otherArgs =

new GenericOptionsParser(conf, args).getRemainingArgs()

if (otherArgs.length != 2) {

System.err.println("Usage: wordcount <in><out>")

System.exit(2)

}

Job job = new Job(conf, "word count")//构建一个job

job.setJarByClass(WordCount.class)

job.setMapperClass(TokenizerMapper.class)

job.setCombinerClass(IntSumReducer.class)

job.setReducerClass(IntSumReducer.class)

job.setOutputKeyClass(Text.class)

job.setOutputValueClass(IntWritable.class)

FileInputFormat.addInputPath(job, new Path(otherArgs[0]))

FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]))

System.exit(job.waitForCompletion(true) ? 0 : 1)

}

（1）Configuration conf=new Configuration()//初始化

运行mapreduce程序前都要初始化Configuration，该类主要是读取mapreduce系统配置信息，这些信息包括hdfs还有mapreduce，也就是安装hadoop时候的配置文件例如：core-site.xml、hdfs-site.xml和mapred-site.xml等等文件里的信息。

程序员开发mapreduce时候只是在填空，在map函数和reduce函数里编写实际进行的业务逻辑，其它的工作都是交给mapreduce框架自己 *** 作的，但是至少我们要告诉它怎么 *** 作啊，比如hdfs在哪里啊，mapreduce的jobstracker在哪里啊，而这些信息就在conf包下的配置文件里。

（2） Job job = new Job(conf, "word count")

在mapreduce框架里一个mapreduce任务也叫mapreduce作业也叫做一个mapreduce的job。

具体的map和reduce运算就是task了，这里我们构建一个job，构建时候有两个参数，一个是conf，一个是这个job的名称。

（3） job.setOutputKeyClass(Text.class)

job.setOutputValueClass(IntWritable.class)

定义输出的key/value的类型，也就是最终存储在hdfs上结果文件的key/value的类型。

（4）

FileInputFormat.addInputPath(job, new Path(otherArgs[0]))

FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]))

System.exit(job.waitForCompletion(true) ? 0 : 1)

第一行就是构建输入的数据文件

第二行是构建输出的数据文件

最后一行是如果job运行成功了，程序就正常退出。

FileInputFormat和FileOutputFormat可以设置输入输出文件路径，

mapreduce计算时候:

输入文件必须存在，不然mapreduce任务直接退出。

输出一般是一个文件夹，而且该文件夹不能存在。

Mapreduce简介

Hadoop MapReduce 源于Google发表的 MapReduce论文。Hadoop MapReduce 其实就是Google MapReduce的一个克隆版本。Hadoop 2.0即第二代Hadoop系统，其框架最核心的设计是HDFS、MapReduce和YARN。其中，HDFS为海量数据提供存储，MapReduce用于分布式计算，YARN用于进行资源管理。

其实，一次mapreduce过程就包括上图的6个步骤，input、splitting、mapping、shuffling、redecing、final redult。

文件要存储在HDFS中，每个文件被切分成多个一定大小的块也就是Block,(Hadoop1.0默认为64M,Hadoop2.0默认为128M),并且默认3个备份存储在多个的节点中。

MR通过Inputformat将数据文件从HDFS中读入取，读取完后会对数据进行split切片，切片的数量根据Block的大小所决定，然后每一个split的个数又决定map的个数，即一个split会分配一个maptask并行实例处理。

如何确定切分的文件大小？

数据进入到map函数中，然后开始按照一定的规则切分。其实这就是我们自定义的计算逻辑，我们编写mr程序的map函数的逻辑一般就在这个阶段执行。企业应用为了追求开发效率，一般都使用hive sql代替繁琐的mr程序了，这里附上一个经典的wordcount的map函数重温一下吧。

Shuffle是我们不需要编写的模块，但却是十分关键的模块。

在map中，每个 map 函数会输出一组 key/value对, Shuffle 阶段需要从所有 map主机上把相同的 key 的 key value对组合在一起，（也就是这里省去的Combiner阶段）组合后传给 reduce主机, 作为输入进入 reduce函数里。

Partitioner组件 负责计算哪些 key 应当被放到同一个 reduce 里

HashPartitioner类，它会把 key 放进一个 hash函数里，然后得到结果。如果两个 key 的哈希值 一样，他们的 key/value对 就被放到同一个 reduce 函数里。我们也把分配到同一个 reduce函数里的 key /value对 叫做一个reduce partition.

我们看到 hash 函数最终产生多少不同的结果, 这个 Hadoop job 就会有多少个 reduce partition／reduce 函数，这些 reduce函数最终被JobTracker 分配到负责 reduce 的主机上，进行处理。

Map方法之后，数据首先进入到分区方法，把数据标记好分区，然后把数据发送到环形缓冲区；环形缓冲区默认大小100m，环形缓冲区达到80%时，进行溢写； 溢写前对数据进行排序 ，排序按照对key的索引进行字典顺序排序，排序的手段快排；溢写产生大量溢写文件，需要 对溢写文件进行归并排序 ；对溢写的文件也可以进行Combiner *** 作，前提是汇总 *** 作，求平均值不行。最后将文件按照分区存储到磁盘，等待Reduce端拉取。每个Reduce拉取Map端对应分区的数据。拉取数据后先存储到内存中，内存不够了，再存储到磁盘。拉取完所有数据后， 采用归并排序将内存和磁盘中的数据都进行排序 。在进入Reduce方法前，可以对数据进行分组 *** 作。值得注意的是， 整个shuffle *** 作是有3次排序的。

reduce() 函数以 key 及对应的 value 列表作为输入，按照用户自己的程序逻辑，经合并 key 相同的 value 值后，产 生另外一系列 key/value 对作为最终输出写入 HDFS。

1.输入:输入数据分为键/值对，由集群中的每个节点处理。2.映射函数:使用输入数据中的每个键/值对来调用用户定义的映射函数，以生成一组中间键/值对。3.Shuffle:将中间的键/值对分组，并将其发送到正确的节点。4.Reduce函数:将同一个键的中间值赋给reduce函数并聚合它们。5.输出:将最终的键/值对发送到输出文件。1)输入数据接口:InputFormat。默认的实现类是:TextInputFormat。TextInputFormat的功能逻辑是:一次读取一行文本，然后以该行的起始偏移量为键，将行内容作为值返回。CombineTextInputFormat可以将多个小文件合并成一个切片，提高处理效率。(2)逻辑处理接口:映射器用户根据业务需求实现三种方法:map() setup() cleanup()。(3)划分器划分HashPartitioner有一个默认实现，逻辑是根据key和numReduces的哈希值返回一个分区号；key.hashCode()&Integer。最大值% numReduces如果业务上有特殊需求，可以定制分区。(4)厘米平行排序当我们使用自定义对象作为输出的键时，必须实现WritableComparable接口，并在其中重写compareTo()方法。部分排序:每个最终输出文件的内部排序。全排序:对所有数据进行排序，通常只减少一次。二次排序:排序有两个条件。(5)合并者合并合并器合并可以提高程序执行的效率，减少IO传输。但使用时不得影响原业务处理结果。(6)逻辑处理接口:减速器用户根据业务需求实现三种方法:reduce() setup() cleanup()。(7)输出数据接口:OutputFormat默认的实现类是TextOutputFormat，功能逻辑是:为每个KV对输出一行到目标文本文件。用户还可以自定义输出格式。

---