[大数据]第三章 MapReduce框架原理

3.1 MapReduce计算模型

MapReduce计算模型主要由三个阶段构成：Map、Shuffle、Reduce。

Map是映射，负责数据的处理、分发，将原始数据转化为键值对；Reduce是合并，将具有相同key值的value进行处理后再输出新的键值对作为最终结果。为了让Reduce可以并行处理Map的结果，必须对Map的输出进行一定的排序与分割，然后再交给对应的Reduce，而这个将Map输出进一步整理并交给Reduce的过程就是Shuffle。整个MR的大致过程如下所示。

Map和Reduce操作需要自己定义相应Mapper类和Reducer类，以完成所需要的化简、合并操作，而shuffle则是系统自动实现的，了解shuffle的具体流程能编写出更加高效的Mapreduce程序。

Shuffle过程包含在Map和Reduce两端，即Map shuffle和Reduce shuffle

3.2 MapReduce工作流程

1.流程示意图

2.流程详解
上面流程是整个MapReduce整个工作流程，但是Shuffle过程只是从第7步开始到第16步结束，具体Shuffle过程详解，如下：
(1) MapTask收集map()方法输出的kv对，放到内存缓冲区中
(2) 从内存缓冲区不断溢出本地磁盘文件，可能会溢出多个文件
(3) 多个溢出文件会被合并成大的溢出文件
(4) 在溢出过程及合并的过程中，都要调用Partitioner进行分区和针对key进行排序
(5) ReduceTask根据自己的分区号，去各个MapTask机器上取相应的结果分区数据
(6) ReduceTask会取到同一个分区的来自不同MapTask的结果文件，ReduceTask会将这些文件再进行合并（归并排序）
(7) 合并成大文件后，Shuffle的过程也就结束了，后面进入ReduceTask的逻辑运算过程（从文件中取出一个一个的键值对Group，调用用户自定义的reduce()方法）

3.注意
Shuffle中的缓冲区大小会影响到MapReduce程序的执行效率，原则上说，缓冲区越大，磁盘io的次数越少，执行速度就越快。

缓冲区的大小可以通过参数调整，参数：mapreduce.task.io.sort.mb 默认100M。

4.源码解析流程

3.3 MapTask工作机制

(1) Read阶段：MapTask通过用户编写的RecordReader，从输入InputSplit中解析出一个个key/value。
(2) Map阶段：该阶段主要是将解析出的key/value交给用户编写map()函数处理，并产生一系列新的key/value。
(3) Collect收集阶段：在用户编写map()函数中，当数据处理完成后，一般会调用OutputCollector.collect()输出结果。在该函数内部，它会将生成的key/value分区（调用Partitioner），并写入一个环形内存缓冲区中。
(4) Spill阶段：即溢写，当环形缓冲区满后，MapReduce会将数据写到本地磁盘上，生成一个临时文件。将数据写入本地磁盘之前，先要对数据进行一次排序（快速排序），并在必要时对数据进行合并、压缩等操作。

溢写阶段详情：
（一）步骤1：利用快速排序算法对环形缓冲区内的数据进行排序，先按照Partition编号进行排序，然后按照key进行排序。经过排序后，数据以分区为单位聚集在一起，且同一分区内所有数据按照key有序。
（二）步骤2：按照分区编号由小到大依次将每个分区中的数据写入临时文件output/spillN.out（N表示当前溢写次数）中。如果用户设置了Combiner，则写入文件之前，对每个分区中的数据进行一次合并操作。
（三）步骤3：将分区数据的元信息写到内存索引数据结构SpillRecord中，其中每个分区的元信息包括在临时文件中的偏移量、压缩前数据大小和压缩后数据大小。如果当前内存索引大小超过1MB，则将内存索引写入到文件output/spillN.out.index中。

(5) Combine阶段：当所有数据处理完成后，MapTask对所有临时文件进行一次合并，以确保最终只会生成一个数据文件，并保存到文件output/file.out中，同时生成相应的索引文件output/file.out.index。

在进行文件合并过程中，MapTask以分区为单位进行合并。对于某个分区，它将采用多轮递归合并的方式。每轮合并mapreduce.task.io.sort.factor（默认10）个文件，并将产生的文件重新加入待合并列表中，对文件排序后（归并排序），重复以上过程，直到最终得到一个大文件。

让每个MapTask最终只生成一个数据文件，可避免同时打开大量文件和同时读取大量小文件产生的随机读取带来的开销。

Map端相关的属性

mapreduce.tasktracker.http.threads与mapreduce.reduce.shuffle.parallelcopies区别
mapreduce.tasktracker.http.threads表示每个TaskTracker能够提供40（默认）个http 线程来服务于reducer（假设reducer的数量有50个，MapTask输出的文件中可能包含50个分区的数据，所以可能有50个reducer同时来拉取数据，该参数限制了拉取数据的reducer数量）
mapreduce.reduce.shuffle.parallelcopies表示一个reducer能够同时从5（默认）个map端获取数据

3.4 Map Shuffle

在Map端的shuffle过程是对Map的结果进行分区、排序、分割，然后将属于同一划分（分区）的输出合并在一起并写在磁盘上，最终得到一个分区有序的文件。分区有序的含义是map输出的键值对按分区进行排列，具有相同partition值的键值对存储在一起，每个分区里面的键值对又按key值进行升序排列（默认），其流程大致如下

3.4.1 Collector环形缓冲区

3.4.1.1.环形缓冲区作用

环形缓冲区分为三块，空闲区、数据区、索引区。初始位置取名叫做“赤道”，就是圆环上的白线那个位置。初始状态的时候，数据和索引都为0，所有空间都是空闲状态。

mapreduce.task.io.sort.mb，默认100M，可以稍微设置大一些，但不要太大，因为每个spilt就128M。

环形缓冲区写入的时候，有个细节：数据是从赤道的右边开始写入，索引（每次申请4kb）是从赤道左边开始写。这个设计很有意思，这样两个文件各是各的，互不干涉。

在数据和索引占用空间到了mapreduce.map.sort.spill.percent参数设置的比例时（默认80%，这个是调优的参数），会有两个动作：
(1) 对写入的数据进行原地排序，并把排序好的数据和索引spill到磁盘上去；
(2) 在空闲的20%区域中，重新计算一个新的赤道，然后在新赤道的右边写入数据，左边写入索引；
(3) 当20%写满了，但是上一次80%的数据还没写到磁盘的时候，程序就会panding一下，等80%空间腾出来之后再继续写。

3.4.1.2.环形缓冲区结构

Map的输出结果是由collector处理的，每个Map任务不断地将键值对输出到内存中构造的一个环形数据结构中。使用环形数据结构是为了更有效地使用内存空间，在内存中放置尽可能多的数据。

环形缓冲区，这个数据结构其实就是个字节数组byte[]，叫Kvbuffer，但是这里面不光放置了数据，还放置了一些索引数据，给放置索引数据的区域起了一个Kvmeta的别名。

数据区域和索引数据区域在Kvbuffer中是相邻不重叠的两个区域，用一个分界点来划分两者，分界点不是亘古不变的，而是每次Spill之后都会更新一次。初始的分界点是0，数据的存储方向是向上增长，索引数据的存储方向是向下增长。

Kvbuffer的指针bufindex（即数据的存储方向）是一直闷着头地向上增长，比如bufindex初始值为0，一个Int型的key写完之后，bufindex增长为4，
一个Int型的value写完之后，bufindex增长为8。（int型的数据占有4个字节）

索引是kvbuffer中的键值对的索引，是个四元组，包括：value的起始位置、key的起始位置、partition值、value的长度，占用四个Int长度，Kvmeta指针Kvindex每次都是向下跳四个“格子”，然后再向上一个格子一个格子地填充四元组的数据。

比如Kvindex初始位置是-4（字节），当第一个键值对写完之后，(Kvindex+0)的位置存放value的起始位置、(Kvindex+1)的位置存放key的起始位置、(Kvindex+2)的位置存放partition的值、(Kvindex+3)的位置存放value的长度，然后Kvindex跳到-8位置，等第二个键值对和索引写完之后，Kvindex跳到-12位置。

关于Spill触发的条件，也就是Kvbuffer用到什么程度开始Spill。如果把Kvbuffer用得一点都不剩的时候再开始Spill，那Map任务就需要等Spill完成腾出空间之后才能继续写数据；如果Kvbuffer只是满到一定程度，比如80%的时候就开始Spill，那在Spill的同时，Map任务还能继续写数据，如果Spill够快，Map可能都不需要为空闲空间而发愁。两利相衡取其大，一般选择后者。Spill的门限可以通过mapreduce.map.sort.spill.percent，默认是0.8。

Spill这个重要的过程是由Spill线程承担，Spill线程从Map任务接到“命令”之后就开始正式干活，干的活叫SortAndSpill，原来不仅仅是Spill，在Spill之前还有个颇具争议性的Sort。

3.4.2 Partition

对于map输出的每一个键值对，系统都会给定一个partition，partition值默认是通过计算key的hash值后对Reduce task的数量取模获得。如果一个键值对的partition值为0，意味着这个键值对会交给第一个Reducer处理。

3.4.3 Sort

当Spill触发后，SortAndSpill先把Kvbuffer中的数据按照partition值和key两个关键字升序排序（快速排序），移动的只是索引数据，排序结果是Kvmeta中数据按照partition为单位聚集在一起，同一partition内的按照key升序。

3.4.4 Spill

Spill线程为这次Spill过程创建一个磁盘文件：从所有的本地目录中轮训查找能存储这么大空间的目录，找到之后在其中创建一个类似于“spill12.out”的文件。Spill线程根据排过序的Kvmeta挨个partition把数据写到这个文件中，一个partition对应的数据写完之后顺序地写下个partition，直到把所有的partition遍历完。一个partition在文件中对应的数据也叫段(segment)。在这个过程中如果用户配置了combiner类，那么在写之前会先调用combineAndSpill()，对结果进行进一步合并后再写出。Combiner会优化MapReduce的中间结果，所以它在整个模型中会多次使用。Combiner的输出是Reducer的输入，Combiner绝不能改变最终的计算结果。

所有的partition对应的数据都放在这个文件里，虽然是顺序存放的，但是怎么直接知道某个partition在这个文件中存放的起始位置呢？强大的索引又出场了。有一个三元组记录某个partition对应的数据在这个文件中的索引：起始位置、原始数据长度、压缩之后的数据长度，一个partition对应一个三元组。然后把这些索引信息存放在内存中，如果内存中放不下了，后续的索引信息就需要写到磁盘文件中：从所有的本地目录中轮训查找能存储这么大空间的目录，找到之后在其中创建一个类似于“spill12.out.index”的文件，文件中不光存储了索引数据，还存储了crc32的校验数据。spill12.out.index不一定在磁盘上创建，如果内存（默认1M空间）中能放得下就放在内存中，即使在磁盘上创建了，和spill12.out文件也不一定在同一个目录下。每一次Spill过程就会最少生成一个out文件，有时还会生成index文件，Spill的次数也烙印在文件名中。索引文件和数据文件的对应关系如下图所示：

在Spill线程如火如荼的进行SortAndSpill工作的同时，Map任务不会因此而停歇，而是一无既往地进行着数据输出。Map还是把数据写到kvbuffer中，那问题就来了：只顾着闷头按照bufindex指针向上增长，kvmeta只顾着按照Kvindex向下增长，是保持指针起始位置不变继续跑呢，还是另谋它路？如果保持指针起始位置不变，很快bufindex和Kvindex就碰头了，碰头之后再重新开始或者移动内存都比较麻烦，不可取。Map取kvbuffer中剩余空间的中间位置，用这个位置设置为新的分界点，bufindex指针移动到这个分界点，Kvindex移动到这个分界点的-16位置，然后两者就可以和谐地按照自己既定的轨迹放置数据了，当Spill完成，空间腾出之后，不需要做任何改动继续前进。

Map任务总要把输出的数据写到磁盘上，即使输出数据量很小在内存中全部能装得下，在最后也会把数据刷到磁盘上。

3.4.5 Merge

Map任务如果输出数据量很大，可能会进行好几次Spill，out文件和Index文件会产生很多，分布在不同的磁盘上。最后把这些文件进行合并的merge过程闪亮登场。

Merge过程怎么知道产生的Spill文件都在哪了呢？从所有的本地目录上扫描得到产生的Spill文件，然后把路径存储在一个数组里。Merge过程又怎么知道Spill的索引信息呢？也是从所有的本地目录上扫描得到Index文件，然后把索引信息存储在一个列表里。在之前Spill过程中的时候为什么不直接把这些信息存储在内存中呢，何必又多了这步扫描的操作？特别是Spill的索引数据，之前当内存超限之后就把数据写到磁盘，现在又要从磁盘把这些数据读出来，装到更多的内存中。之所以多此一举，是因为这时kvbuffer这个内存大户已经不再使用了，可以回收，因此有内存空间来装这些数据了。（对于内存空间较大的土豪来说，用内存来省却这两个io步骤还是值得考虑的。）

然后为merge过程创建一个叫file.out的文件和一个叫file.out.Index的文件用来存储最终的输出和索引，一个partition一个partition的进行合并输出。对于某个partition来说，从索引列表中查询这个partition对应的所有索引信息，每个对应一个段插入到段列表中。也就是这个partition对应一个段列表，记录所有的Spill文件中对应的这个partition那段数据的文件名、起始位置、长度等等。

然后对这个partition对应的所有的segment进行合并，目标是合并成一个segment。当这个partition对应很多个segment时，会分批地进行合并：先从segment列表中把第一批取出来，以key为关键字放置成最小堆，然后从最小堆中每次取出最小的输出到一个临时文件中，这样就把这一批段合并成一个临时的段，把它加回到segment列表中；再从segment列表中把第二批取出来合并输出到一个临时segment，把其加入到列表中；这样往复执行，直到剩下的段是一批，输出到最终的文件中。最终的索引数据仍然输出到Index文件中。

3.5 Reduce Shuffle

3.5.1 ReduceTask工作机制

(1) Copy阶段
由于Job的每一个map都会根据reduce(n)数将数据分成n个partition，所以map的中间结果有可能包含每一个reduce需要处理的部分数据的。所以，为了优化reduce的执行时间，hadoop等Job的第一个map结束后，所有的reduce就开始尝试从完成的map下载该reduce对应的partition部分数据，因此map和reduce是交叉进行的，其实就是shuffle。Reduce任务通过HTTP向各个Map任务拖取（下载）它所需要的数据（网络传输），Reducer是如何知道要去哪些机器取数据呢？一旦map任务完成之后，就会通过常规心跳通知父TaskTracker状态已经更新，TaskTracker进而通知JobTracker（这些通知在心跳机制中进行）。reduce的一个线程会周期性地向JobTracker询问，直到提取完所有数据，数据被reduce提走之后，map机器不会立刻删除数据，这是为了预防reduce任务失败需要重做。因此map输出数据是在整个作业完成之后才被删除掉的。

reduce进程启动数据copy线程(Fetcher)，通过HTTP方式请求map task所在的TaskTracker获取map task的输出文件。由于map通常有许多个，所以对一个reduce来说，下载也可以是并行的从多个map下载，那到底同时到多少个Mapper下载数据？这个并行度是可以通过mapreduce.reduce.shuffle.parallelcopies(default5）调整。默认情况下，每个Reducer只会有5个map端并行的下载线程在从map下数据，如果一个时间段内Job完成的map有100个或者更多，那么reduce也最多只能同时下载5个map的数据，所以这个参数比较适合map很多并且完成比较快的情况下调大，有利于reduce更快的获取属于自己部分的数据。在Reducer内存和网络都比较好的情况下，可以调大该参数；

reduce的每一个下载线程在下载某个map数据的时候，有可能因为那个map中间结果所在机器发生错误，或者中间结果的文件丢失，或者网络瞬断等等情况，这样reduce的下载就有可能失败，所以reduce的下载线程并不会无休止的等待下去，当一定时间后下载仍然失败，那么下载线程就会放弃这次下载，并在随后尝试从另外的地方下载（因为这段时间map可能重跑）。reduce下载线程的这个最大的下载时间段是可以通过mapreduce.reduce.shuffle.read.timeout（default180000毫秒）调整的。如果集群环境的网络本身是瓶颈，那么用户可以通过调大这个参数来避免reduce下载线程被误判为失败的情况。一般情况下都会调大这个参数，这是企业级最佳实战。

(2) Merge Sort阶段
这里的merge和map端的merge动作类似，只是缓冲区中存放的是不同map端copy来的数值。Copy过来的数据会先放入内存缓冲区中，然后当内存缓冲区的使用量达到一定程度的时候才spill磁盘。这里的缓冲区大小要比map端的更为灵活，它基于JVM的heap size设置。这个内存大小的控制不像map一样可以通过mapreduce.map.sort.spill.percent来设定，而是通过另外一个参数 mapreduce.reduce.shuffle.input.buffer.percent（default 0.7f 源码里面写死了） 来设置，这个参数其实是一个百分比，即reduce的shuffle阶段内存缓冲区，最多使用量为：0.7 × maxHeap of reduce task。JVM的heapsize的70%。内存到磁盘merge的启动门限可以通过mapreduce.reduce.shuffle.merge.percent（default 0.66）配置。也就是说，如果该reduce task的最大heap使用量（通常通过mapreduce.admin.reduce.child.java.opts来设置，比如设置为-Xmx1024m）的一定比例用来缓存数据。默认情况下，reduce会使用其heap size的70%来在内存中缓存数据。假设 mapreduce.reduce.shuffle.input.buffer.percent 为0.7，reduce task的max heapsize为1G，那么用来做下载数据缓存的内存就为大概700MB左右。这700M的内存，跟map端一样，也不是要等到全部写满才会往磁盘刷的，而是当这700M中被使用到了一定的限度（通常是一个百分比），就会开始往磁盘刷（刷磁盘前会先做sort merge）。这个限度阈值也是可以通过参数 mapreduce.reduce.shuffle.merge.percent（default 0.66）来设定。与map 端类似，这也是溢写的过程，这个过程中如果设置了Combiner，也是会启用的，然后在磁盘中生成了众多的溢写文件。这种merge方式一直在运行，直到没有map端的数据时才结束，然后启动磁盘到磁盘的merge方式生成最终的那个文件，merge的时候会进行归并排序。一般Reduce是一边copy一边sort，即copy和sort两个阶段是重叠而不是完全分开的。最终Reduce shuffle过程会输出一个整体有序的数据块。

merge有三种形式
(1) 内存到内存（memToMemMerger）
Hadoop定义了一种MemToMem合并，这种合并将内存中的map输出合并，然后再写入内存。这种合并默认关闭，可以通过mapreduce.reduce.merge.memtomem.enabled(default:false)打开，当map输出文件达到mapreduce.reduce.merge.memtomem.threshold时，触发这种合并。

(2) 内存中Merge（inMemoryMerger）/内存到磁盘
当缓存区数据达到配置的阈值时，这些数据在内存中被合并、写入机器磁盘。阈值有2种配置方式：

• 配置内存比例：前面提到reduce JVM堆内存的一部分用于存放来自map任务的输出，在这基础之上配置一个开始合并数据的比例。假设用于存放map输出的内存为500M，mapreduce.reduce.shuffle.merge.percent配置为0.66，则当内存中的数据达到330M的时候，会触发合并写入。

• 配置map输出数量：通过mapreduce.reduce.merge.inmem.threshold配置。在合并的过程中，会对被合并的文件做全局的排序。如果作业配置了Combiner，则会运行combine函数，减少写入磁盘的数据量。

(3) 磁盘上的Merge（onDiskMerger）
在copy过来的数据不断写入磁盘的过程中，一个后台线程会把这些文件合并为更大的、有序的文件。如果map的输出结果进行了压缩，则在合并过程中，需要在内存中解压后才能进行合并。这里的合并只是为了减少最终合并的工作量，也就是还在拷贝map输出时，就开始进行一部分合并工作。合并的过程一样会进行全局排序（归并排序）。

最终磁盘中Merge，当所有map输出都拷贝完毕之后，所有数据被最后合并成一个整体有序的文件，作为reduce任务的输入。这个合并过程是一轮一轮进行的，最后一轮的合并结果直接推送给reduce作为输入，节省了磁盘操作的一个来回。最后（所有map输出都拷贝到reduce之后）进行合并的map输出可能来自合并后写入磁盘的文件，也可能来及内存缓冲，在最后写入内存的map输出可能没有达到阈值触发合并，所以还留在内存中。

每一轮合并不一定合并平均数量的文件数，指导原则是使整个合并过程中写入磁盘的数据量最小，为了达到这个目的，则需要最终的一轮合并中合并尽可能多的数据，因为最后一轮的数据直接作为reduce的输入，无需写入磁盘再读出。因此让最终的一轮合并的文件数达到最大，即合并因子的值，通过mapreduce.task.io.sort.factor（default：10）来配置。

(3) Reduce阶段
当reducer将所有的map上对应自己partition的数据下载完成后，就会开始真正的reduce计算阶段。由于reduce计算时肯定也是需要消耗内存的，而在读取reduce需要的数据时，同样是需要内存作为buffer，reduce需要多少的内存百分比来作为reducer读已经sort好的数据的buffer大小？？默认情况下为0，也就是说，默认情况下，reduce是全部从磁盘开始读处理数据。可以用mapreduce.reduce.input.buffer.percent（default 0.0）(源代码MergeManagerImpl.java：674行)来设置reduce的缓存。如果这个参数大于0，那么就会有一定量的数据被缓存在内存并输送给reduce，当reduce计算逻辑消耗内存很小时，可以分一部分内存用来缓存数据，可以提升计算的速度。所以默认情况下都是从磁盘读取数据，如果内存足够大的话，务必设置该参数让reduce直接从缓存读数据，这样做就有点Spark Cache的感觉；

Reduce阶段，框架为已分组的输入数据中的每个 <key, (list of values)>对调用一次 reduce(WritableComparable,Iterator, OutputCollector, Reporter)方法。Reduce任务的输出通常是通过调用 OutputCollector.collect(WritableComparable,Writable)写入文件系统的。Reducer的输出是没有排序的。

Reduce端相关属性

3.5.2 设置ReduceTask并行度（个数）

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

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

1.实验：测试ReduceTask多少合适
(1) 实验环境：1个Master节点，16个Slave节点：CPU 8GHZ，内存 2G
(2) 实验结论：
改变ReduceTask （数据量为1GB）
MapTask =16

2.注意事项
(1) ReduceTask=0，表示没有Reduce阶段，输出文件个数和Map个数一致；
(2) ReduceTask默认值为1，所以输出文件个数为1个；
(3) 如果数据分布不均匀，就有可能在Reduce阶段产生数据倾斜；
(4) ReduceTask数量并不是任意设置，还要考虑业务逻辑需求，有些情况下，需要计算全局汇总结果，就只能有1个ReduceTask；
(5) 具体多少个ReduceTask，需要根据集群性能而定；
(6) 如果分区数不是1，但是ReduceTask为1，是否执行分区过程。答案是：不执行分区过程。因为MapTask源码中，执行分区的前提是先判断ReduceNum个数是否大于1，不大于1肯定不执行。

3.5.3 mapreduce中partition数量与reduce task数量对结果影响

(1) 情况1：Partition数量为1，reduce数量为3。
设置reduce数量为3

Job.setNumReduceTasks(3);

设置partition数量为1，返回值为0

public class MyPartitioner extends HashPartitioner<MyKey, DoubleWritable> {
	// 执行时间越短越好
	public int getPartition(MyKey key, DoubleWritable value, int numReduceTasks){
		// return (key.getYear()-1949)%numReduceTasks;
		  return 0;
	}
}

结果输出3个文件，但只有part-r-00000有内容，其它两个为0字节。

(2) 情况2：Partition数量为1，返回值5，reduce数量为3。
设置reduce数量为3

Job.setNumReduceTasks(3);

设置partition数量为1，返回值为5

public class MyPartitioner extends HashPartitioner<MyKey, DoubleWritable> {
	// 执行时间越短越好
	public int getPartition(MyKey key, DoubleWritable value, int numReduceTasks){
		// return (key.getYear()-1949)%numReduceTasks;
		  return 5;
	}
}

结果输出3个文件，都为0字节。

(3) 情况3：Partition数量为1，返回值1，reduce数量为3。
结果为输出3个文件，但只有part-r-00001有值。

(4) 情况4：Partition数量为3，返回值0、1、2，reduce数量为2。
如果1<ReduceTask的数量<partition数，则有一部分分区数据无处安放，会Exception；如果ReduceTask的数量=1，则不管MapTask端输出多少个分区文件，最终结果都交给这一个ReduceTask，最终也就只会产生一个结果文件 part-r-00000

总结：生成文件数量由reduce数量决定，值输出到哪个文件由partition返回值决定，输出值正确性由partition数量和返回值决定。一般情况是有多少个partition对应多少个reduce。

3.6 InputFormat数据输入

3.6.1切片与MapTask并行度决定机制
1.问题引出
MapTask的并行度决定Map阶段任务处理并发度，进而影响到整个Job的处理速度。

思考：1G的数据，启动8个MapTask，可以提高集群的并发处理能力。那么1K的数据，也启动8个MapTask，会提高集群性能吗？MapTask并行任务是否越多越好呢？哪些因素影响了MapTask并行度？

2.MapTask并行度决定机制
数据块：Block是HDFS物理上把数据分成一块一块。
数据切片：数据切片只是在逻辑上对输入进行分片，并不会在磁盘上将其切分成片进行存储。

3.6.2 Job提交流程源码和切片源码详解

1.Job提交流程源码详解

waitForCompletion()

submit();

// 1建立连接
	connect();	
		// 1）创建提交Job的代理
		new Cluster(getConfiguration());
			// （1）判断是本地yarn还是远程
			initialize(jobTrackAddr, conf); 

// 2 提交job
submitter.submitJobInternal(Job.this, cluster)
	// 1）创建给集群提交数据的Stag路径
	Path jobStagingArea = JobSubmissionFiles.getStagingDir(cluster, conf);

	// 2）获取jobid，并创建Job路径
	JobID jobId = submitClient.getNewJobID();

	// 3）拷贝jar包到集群
    copyAndConfigureFiles(job, submitJobDir);	
	rUploader.uploadFiles(job, jobSubmitDir);

    // 4）计算切片，生成切片规划文件
    writeSplits(job, submitJobDir);
		maps = writeNewSplits(job, jobSubmitDir);
		input.getSplits(job);

    // 5）向Stag路径写XML配置文件
    writeConf(conf, submitJobFile);
	conf.writeXml(out);

    // 6）提交Job,返回提交状态
    status = submitClient.submitJob(jobId, submitJobDir.toString(), job.getCredentials());

2.FileInputFormat切片源码解析(input.getSplits(job))
(1) 程序先找到数据存储的目录；
(2) 开始遍历处理（规划切片）目录下的每一个文件；
(3) 遍历第一个文件ss.txt；
? a.获取文件大小fs.sizeOf(ss.txt)
? b.计算切片大小

computeSplitSize(Math.max(minSize, Math.min(maxSize,blocksize)))=blocksize=128M

? c.默认情况下，切片大小=blocksize
? d.开始切片，形成第一个切片：ss.txt-0:128M，第二个切片ss.txt-128M:256M，第三个切片ss.txt-256M:300M （每次切片时，都要判断切完剩下的部分是否大于块的1.1倍，不大于1.1倍就划分一块切片）
? e.将切片信息写到一个切片规划文件中；
? f.整个切片的核心过程在getSplit()方法中完成；
? g.InputSplit只记录了切片的元数据信息，比如起始位置、长度以及所在的节点列表等。

(4) 提交切片规划文件到YARN上，YARN上的MrAppMaster就可以根据切片规划文件计算开启MapTask数量

3.6.3 FileInputFormat切片机制

切片机制

• 简单地按照文件的内容长度进行切片；
• 切片大小，默认等于Block大小；
• 切片时不考虑数据集整体，而是逐个针对每一个文件单独切片；

案例分析
(1) 输入数据有两个文件

file1.txt    320M
file2.txt    10M

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

file1.txt.split1    0~128
fle1.txt.split2    128~256
file1.txt.split3    256~300
file2.txt.split1    0~10

源码中计算切片大小的公式

Math.max(minSize, Math.min(maxSize, blockSize))

mapreduce.input.fileinputformat.split.minsize默认值为1
mapreduce.input.fileinputformat.split.maxsize默认值为Long.MAXValue，所以默认情况下，切片大小=blockSize

切片大小设置

• maxSize（切片最大值）：如果该参数调得比blockSize小，则会让切片变小，而且就等于配置的这个参数得值；
• minSize（切片最小值）：如果该参数调得比blockSize大，则可以让切片变得比blockSize大；

获取切片信息API

// 获取切片的文件名称
String name = inputSplit.getPath().getName();
// 根据文件类型，获取切片信息
FileSplit inputSplit = (FileSplit) context.getInputSplit();

3.6.4 CombineTextInputFormat切片机制

框架默认的TextInputFormat切片机制是对任务按文件规划切片，不管文件多小，都会是一个单独的切片，都会交给一个MapTask，这样如果有大量小文件，就会产生大量的MapTask，处理效率极其低下。

1.应用场景
CombineTextInputFormat用于小文件过多的场景，它可以将多个小文件从逻辑上规划到一个切片中，这样，多个小文件就可以交给一个MapTask处理。

2.虚拟存储切片最大值设置
CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);// 4m
注意：虚拟存储切片最大值设置最好根据实际的小文件大小情况来设置具体的值。

3.切片机制
生成切片过程包括：虚拟存储过程和切片过程二部分。
(1) 虚拟存储过程
将输入目录下所有文件大小，依次和设置的setMaxInputSplitSize值比较，如果不大于设置的最大值，逻辑上划分一个块。如果输入文件大于设置的最大值且大于两倍，那么以最大值切割一块；当剩余数据大小超过设置的最大值且不大于最大值2倍，此时将文件均分成2个虚拟存储块（防止出现太小切片）。

例如setMaxInputSplitSize值为4M，输入文件大小为8.02M，则先逻辑上分成一个4M。剩余的大小为4.02M，如果按照4M逻辑划分，就会出现0.02M的小的虚拟存储文件，所以将剩余的4.02M文件切分成（2.01M和2.01M）两个文件。

(2) 切片过程
?（a）判断虚拟存储的文件大小是否大于setMaxInputSplitSize值，大于等于则单独形成一个切片。
?（b）如果不大于则跟下一个虚拟存储文件进行合并，共同形成一个切片。
?（c）测试举例：有4个小文件大小分别为1.7M、5.1M、3.4M以及6.8M这四个小文件，则虚拟存储之后形成6个文件块，大小分别为：
1.7M，（2.55M、2.55M），3.4M以及（3.4M、3.4M）
最终会形成3个切片，大小分别为：
（1.7+2.55）M，（2.55+3.4）M，（3.4+3.4）M

3.6.5 CombineTextInputFormat案例

1.需求
将输入的大量小文件合并成一个切片统一处理。

(1) 输入数据
准备4个小文件

(2) 期望
期望一个切片处理4个文件

2.实现过程
(1) 不做任何处理，运行1.8节的WordCount案例程序，观察切片个数为4。

number of splits: 4

(2) 在WordcountDriver中增加如下代码，运行程序，并观察运行的切片个数为3。
驱动类中添加代码如下

// 如果不设置InputFormat，它默认用的是TextInputFormat.class
job.setInputFormatClass(CombineTextInputFormat.class);

//虚拟存储切片最大值设置4m
CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);

运行如果为3个切片。

number of splits: 3

(3) 在WordcountDriver中增加如下代码，运行程序，并观察运行的切片个数为1。
驱动中添加代码如下

// 如果不设置InputFormat，它默认用的是TextInputFormat.class
job.setInputFormatClass(CombineTextInputFormat.class);

//虚拟存储切片最大值设置20m
CombineTextInputFormat.setMaxInputSplitSize(job, 20971520);

运行如果为1个切片。

number of splits: 1

3.6.6 FileInputFormat实现类

思考：在运行MapReduce程序时，输入的文件格式包括：基于行的日志文件、二进制格式文件、数据库表等。那么，针对不同的数据类型，MapReduce是如何读取这些数据了？

FileInputFormat常见的接口实现类包括：TextInputFormat、KeyValueTextInputFormat、NLineInputFormat、CombineTextInputFormat和自定义InputFormat等。

1.TextInputFormat

TextInputFormat是默认的FileInputFormat实现类，按行读取每条记录。

• 键key是存储该行在整个文件中的起始字节偏移量， LongWritable类型。
• 值value是这行的内容，不包括任何行终止符（换行符和回车符），Text类型。

比如，一个分片包含了如下4条文本记录。

Rich learning form
Intelligent learning engine
Learning more convenient
From the real demand for more close to the enterprise

每条记录表示为以下键/值对

(0,Rich learning form)
(19,Intelligent learning engine)
(47,Learning more convenient)
(72,From the real demand for more close to the enterprise)

2.KeyValueTextInputFormat
每一行均为一条记录，被分隔符分割为key，value。可以通过在驱动类中设置来设定分隔符

conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPERATOR, "\t")

默认分隔符是tab（\t）。

例如，输入是一个包含4条记录的分片。其中——>表示一个（水平方向的）制表符。

line1 ——>Rich learning form
line2 ——>Intelligent learning engine
line3 ——>Learning more convenient
line4 ——>From the real demand for more close to the enterprise

每条记录表示为以下键值对

(line1,Rich learning form)
(line2,Intelligent learning engine)
(line3,Learning more convenient)
(line4,From the real demand for more close to the enterprise)

此时的键key是每行排在制表符之前的Text序列。

3.NLineInputFormat
如果使用NlineInputFormat，代表每个map进程处理的InputSplit不再按Block块去划分，而是按NlineInputFormat指定的行数N来划分。即输入文件的总行数/N=切片数，如果不整除，切片数=商+1。

例如，以上面的4行输入为例

Rich learning form
Intelligent learning engine
Learning more convenient
From the real demand for more close to the enterprise

如果N是2，则每个输入分片包含两行，开启2个MapTask。

(0,Rich learning form)
(19,Intelligent learning engine)

另一个 mapper 则收到后两行

(47,Learning more convenient)
(72,From the real demand for more close to the enterprise)

这里的键和值与TextInputFormat生成的一样。

3.6.7 KeyValueTextInputFormat使用案例

1.需求
统计输入文件中每一行的第一个单词相同的行数。

(1) 输入数据

banzhang ni hao
xihuan hadoop banzhang
banzhang ni hao
xihuan hadoop banzhang

(2) 期望结果数据

banzhang	2
xihuan	2

2.需求分析

3.代码实现
(1) 编写Mapper类

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

public class KVTextMapper extends Mapper<Text, Text, Text, LongWritable>{
	
    // 1 设置value
    LongWritable v = new LongWritable(1);  
    
	@Override
	protected void map(Text key, Text value, Context context)
			throws IOException, InterruptedException {

    // banzhang ni hao
        
    // 2 写出
    context.write(key, v);  
	}
}

(2) 编写Reducer类

package com.test.mapreduce.KeyValueTextInputFormat;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class KVTextReducer extends Reducer<Text, LongWritable, Text, LongWritable>{
	
    LongWritable v = new LongWritable();  
    
	@Override
	protected void reduce(Text key, Iterable<LongWritable> values,	Context context) throws IOException, InterruptedException {
		
		 long sum = 0L;  

		 // 1 汇总统计
        for (LongWritable value : values) {  
            sum += value.get();  
        }
         
        v.set(sum);  
         
        // 2 输出
        context.write(key, v);  
	}
}

(3) 编写Driver类

package com.test.mapreduce.keyvaleTextInputFormat;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
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.input.KeyValueLineRecordReader;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class KVTextDriver {

	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		
		Configuration conf = new Configuration();
		// 设置切割符
	    conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPERATOR, " ");
		// 1 获取job对象
		Job job = Job.getInstance(conf);
		
		// 2 设置Driver，关联mapper和reducer
		job.setJarByClass(KVTextDriver.class);
		job.setMapperClass(KVTextMapper.class);
        job.setReducerClass(KVTextReducer.class);
				
		// 3 设置map输出kv类型
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(LongWritable.class);

		// 4 设置最终输出kv类型
		job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);
		
		// 5 设置输入输出数据路径
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		
		// 设置输入格式
	    job.setInputFormatClass(KeyValueTextInputFormat.class);
		
		// 6 设置输出数据路径
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		
		// 7 提交job
		job.waitForCompletion(true);
	}
}

3.6.8 NLineInputFormat使用案例

1.需求
对每个单词进行个数统计，根据每个输入文件的行数来规定输出多少个切片。此案例要求每三行放入一个切片中。
(1) 输入数据

banzhang ni hao
xihuan hadoop banzhang
banzhang ni hao
xihuan hadoop banzhang
banzhang ni hao
xihuan hadoop banzhang
banzhang ni hao
xihuan hadoop banzhang
banzhang ni hao
xihuan hadoop banzhang banzhang ni hao
xihuan hadoop banzhang

(2) 期望输出数据

Number of splits:4

2.需求分析

3.代码实现
(1) 编写Mapper类

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

public class NLineMapper extends Mapper<LongWritable, Text, Text, LongWritable>{
	
	private Text k = new Text();
	private LongWritable v = new LongWritable(1);
	
	@Override
	protected void map(LongWritable key, Text value, Context context)	throws IOException, InterruptedException {
		
		 // 1 获取一行
        String line = value.toString();
        
        // 2 切割
        String[] splited = line.split(" ");
        
        // 3 循环写出
        for (int i = 0; i < splited.length; i++) {
            k.set(splited[i]);
            context.write(k, v);
        }
	}
}

(2) 编写Reducer类

package com.test.mapreduce.nline;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class NLineReducer extends Reducer<Text, LongWritable, Text, LongWritable>{
	
	LongWritable v = new LongWritable();
	
	@Override
	protected void reduce(Text key, Iterable<LongWritable> values,	Context context) throws IOException, InterruptedException {
		
        long sum = 0l;

        // 1 汇总
        for (LongWritable value : values) {
            sum += value.get();
        }  
        
        v.set(sum);
        
        // 2 输出
        context.write(key, v);
	}
}

(3) 编写Driver类

package com.test.mapreduce.nline;
import java.io.IOException;
import java.net.URISyntaxException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
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.input.NLineInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class NLineDriver {
	
	public static void main(String[] args) throws IOException, URISyntaxException, ClassNotFoundException, InterruptedException {
		
// 输入输出路径需要根据自己电脑上实际的输入输出路径设置
args = new String[] { "e:/input/inputword", "e:/output1" };

		// 1 获取job对象
		Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);
        
        // 7设置每个切片InputSplit中划分三条记录
        NLineInputFormat.setNumLinesPerSplit(job, 3);
          
        // 8使用NLineInputFormat处理记录数  
        job.setInputFormatClass(NLineInputFormat.class);  
          
        // 2设置Driver，关联mapper和reducer
        job.setJarByClass(NLineDriver.class);  
        job.setMapperClass(NLineMapper.class);  
        job.setReducerClass(NLineReducer.class);  
        
        // 3设置map输出kv类型
        job.setMapOutputKeyClass(Text.class);  
        job.setMapOutputValueClass(LongWritable.class);  
        
        // 4设置最终输出kv类型
        job.setOutputKeyClass(Text.class);  
        job.setOutputValueClass(LongWritable.class);  
          
        // 5设置输入输出数据路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));  
        FileOutputFormat.setOutputPath(job, new Path(args[1]));  
          
        // 6提交job
        job.waitForCompletion(true);  
	}
}

4.测试
(1) 输入数据

banzhang ni hao
xihuan hadoop banzhang
banzhang ni hao
xihuan hadoop banzhang
banzhang ni hao
xihuan hadoop banzhang
banzhang ni hao
xihuan hadoop banzhang
banzhang ni hao
xihuan hadoop banzhang banzhang ni hao
xihuan hadoop banzhang

(2)输出结果的切片数，如图所示

3.6.9 自定义InputFormat

在企业开发中，Hadoop框架自带的InputFormat类型不能满足所有应用场景，需要自定义InputFormat来解决实际问题。

自定义InputFormat步骤如下

1. 自定义一个类继承FileInputFormat；
2. 改写RecordReader，实现一次读取一个完整文件封装为KV；
3. 在输出时使用SequenceFileOutPutFormat输出合并文件；

3.6.10 自定义InputFormat案例

无论HDFS还是MapReduce，在处理小文件时效率都非常低，但又难免面临处理大量小文件的场景，此时，就需要有相应解决方案。可以自定义InputFormat实现小文件的合并。

1.需求
将多个小文件合并成一个SequenceFile文件（SequenceFile文件是Hadoop用来存储二进制形式的key-value对的文件格式），SequenceFile里面存储着多个文件，存储的形式为文件路径+名称为key，文件内容为value。

(1) 输入数据

one.txt    two.txt    three.txt

(2) 期望输出文件格式

part-r-00000

2.需求分析
自定义一个类继承FileInputFormat

• 重写isSplitable()方法，返回false不可切割；
• 重写createRecordReader()，创建自定义的RecordReader对象，并初始化；

改写RecordReader，实现一次读取一个完整文件封装为KV

• 采用IO流一次读取一个文件输出到value中，因为设置了不可能分片，最终把所有文件都封装到了value中；
• 获取文件路径+名称，并设置key；

设置Driver

// 设置输入的InputForMat
job.setInputFormatClass(WholeFileInputFormat.class);
// 设置输出的OutputFormat
job.setOutputFormatClass(SequenceFileOutputFormat.class);

3.程序实现
(1) 自定义InputFromat

package com.test.mapreduce.inputformat;
import java.io.IOException;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

// 定义类继承FileInputFormat
public class WholeFileInputformat extends FileInputFormat<Text, BytesWritable>{
	
	@Override
	protected boolean isSplitable(JobContext context, Path filename) {
		return false;
	}

	@Override
	public RecordReader<Text, BytesWritable> createRecordReader(InputSplit split, TaskAttemptContext context)	throws IOException, InterruptedException {
		WholeRecordReader recordReader = new WholeRecordReader();
		recordReader.initialize(split, context);
		return recordReader;
	}
}

(2) 自定义RecordReader类

package com.test.mapreduce.inputformat;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

public class WholeRecordReader extends RecordReader<Text, BytesWritable>{

	private Configuration configuration;
	private FileSplit split;
	
	private boolean isProgress= true;
	private BytesWritable value = new BytesWritable();
	private Text k = new Text();

	@Override
	public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
		this.split = (FileSplit)split;
		configuration = context.getConfiguration();
	}

	@Override
	public boolean nextKeyValue() throws IOException, InterruptedException {
		
		if (isProgress) {
			// 1 定义缓存区
			byte[] contents = new byte[(int)split.getLength()];
			
			FileSystem fs = null;
			FSDataInputStream fis = null;
			
			try {
				// 2 获取文件系统
				Path path = split.getPath();
				fs = path.getFileSystem(configuration);
				
				// 3 读取数据
				fis = fs.open(path);
				
				// 4 读取文件内容
				IOUtils.readFully(fis, contents, 0, contents.length);
				
				// 5 输出文件内容
				value.set(contents, 0, contents.length);

               // 6 获取文件路径及名称
               String name = split.getPath().toString();

              // 7 设置输出的key值
              k.set(name);

			} catch (Exception e) {
				
			}finally {
				IOUtils.closeStream(fis);
			}
			
			isProgress = false;
			
			return true;
		}
		
		return false;
	}

	@Override
	public Text getCurrentKey() throws IOException, InterruptedException {
		return k;
	}

	@Override
	public BytesWritable getCurrentValue() throws IOException, InterruptedException {
	    return value;
	}

	@Override
	public float getProgress() throws IOException, InterruptedException {
	    return 0;
	}

	@Override
	public void close() throws IOException {
	}
}

(3) 编写SequenceFileMapper类处理流程

package com.test.mapreduce.inputformat;
import java.io.IOException;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

public class SequenceFileMapper extends Mapper<Text, BytesWritable, Text, BytesWritable>{
	
	@Override
	protected void map(Text key, BytesWritable value,			Context context)		throws IOException, InterruptedException {
		context.write(key, value);
	}
}

(4) 编写SequenceFileReducer类处理流程

package com.test.mapreduce.inputformat;
import java.io.IOException;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class SequenceFileReducer extends Reducer<Text, BytesWritable, Text, BytesWritable> {

	@Override
	protected void reduce(Text key, Iterable<BytesWritable> values, Context context)		throws IOException, InterruptedException {

		context.write(key, values.iterator().next());
	}
}
(5)编写SequenceFileDriver类处理流程
package com.test.mapreduce.inputformat;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
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 org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;

public class SequenceFileDriver {

	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		
       // 输入输出路径需要根据自己电脑上实际的输入输出路径设置
		args = new String[] { "e:/input/inputformat", "e:/output1" };

       // 1 获取job对象
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);

       // 2 设置jar包存储位置、关联自定义的mapper和reducer
		job.setJarByClass(SequenceFileDriver.class);
		job.setMapperClass(SequenceFileMapper.class);
		job.setReducerClass(SequenceFileReducer.class);

        // 7设置输入的inputFormat
		job.setInputFormatClass(WholeFileInputformat.class);

        // 8设置输出的outputFormat
	    job.setOutputFormatClass(SequenceFileOutputFormat.class);
       
        // 3 设置map输出端的kv类型
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(BytesWritable.class);
		
       // 4 设置最终输出端的kv类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(BytesWritable.class);

       // 5 设置输入输出路径
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

       // 6 提交job
		boolean result = job.waitForCompletion(true);
		System.exit(result ? 0 : 1);
	}
}

3.7 Shuffle机制

3.7.1 Shuffle机制

Map方法之后，Reduce方法之前的数据处理过程称之为Shuffle，如图所示。

3.7.2 Partition分区

为什么要分区？
要求将统计结果按照条件输出到不同文件中（分区）。比如：将统计结果按
照手机归属地属于不同省份输出到不同文件中（分区）

默认 partition 分区

/**
	源码中：numReduceTasks如果等于1 不会走getPartition方法
	numReduceTasks：默认是1
*/
public class HashPartitioner<K, V> extends Partitioner<K, V> {
	public int getPartition(K key, V value, int numReduceTasks) {
		return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
	}
}

默认分区是根据 key 的 hashCode 对 reduceTasks 个数取模得到的。用户没法控制哪个key 存储到哪个分区

自定义分区
自定义类继承 Partitioner，重写 getPartition()方法。

/**
	该方法返回不同的partition的值，从而就控制了分区个数、前提是numReduceTasks不等于1
	KV:是Map的输出
*/
public class ProvincePartitioner extends Partitioner<Text, FlowBean> {
	@Override
	public int getPartition(Text key, FlowBean value, int numPartitions) {
		// 1 获取电话号码的前三位
		String preNum = key.toString().substring(0, 3);
		int partition = 4;
		// 2 判断是哪个省
		if ("136".equals(preNum)) {
		    partition = 0;
		}else if ("137".equals(preNum)) {
			partition = 1;
		}else if ("138".equals(preNum)) {
			partition = 2;
		}else if ("139".equals(preNum)) {
			partition = 3;
		}
		return partition;
	}
}

在Driver中设置partition

job.setPartitionerClass(CustomPartitioner.class);

自定义 partition 后，要根据自定义 partitioner 的逻辑设置相应数量的 reduce task

job.setNumReduceTasks(5);

总结

• 如果 reduceTask 的数量 > getPartition 的结果数，则会多产生几个空的输出文件part-r-000xx；
• 如果 1 < reduceTask的数量 < getPartition 的结果数，则有一部分分区数据无处安放，会Exception；
• 如果 reduceTask 的数量 = 1，则不管 mapTask 端输出多少个分区文件，最终结果都交给这一个 reduceTask，最终也就只会产生一个结果文件 part-r-00000；

例如：假设自定义分区数为 5，则
(1) job.setNumReduceTasks(1);会正常运行，只不过会产生一个输出文件
(2) job.setNumReduceTasks(2);会报错
(3) job.setNumReduceTasks(6);大于 5，程序会正常运行，会产生空文件

3.7.3 Partition分区案例

1.需求
将统计结果按照手机归属地不同省份输出到不同文件中（分区）
(1) 输入数据phone_data.txt

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 	13729199489	192.168.100.6			240	0	200
10 	13630577991	192.168.100.7	www.shouhu.com	6960	690	200
11 	15043685818	192.168.100.8	www.baidu.com	3659	3538	200
12 	15959002129	192.168.100.9	www.atguigu.com	1938	180	500
13 	13560439638	192.168.100.10			918	4938	200
14 	13470253144	192.168.100.11			180	180	200
15 	13682846555	192.168.100.12	www.qq.com	1938	2910	200
16 	13992314666	192.168.100.13	www.gaga.com	3008	3720	200
17 	13509468723	192.168.100.14	www.qinghua.com	7335	110349	404
18 	18390173782	192.168.100.15	www.sogou.com	9531	2412	200
19 	13975057813	192.168.100.16	www.baidu.com	11058	48243	200
20 	13768778790	192.168.100.17			120	120	200
21 	13568436656	192.168.100.18	www.alibaba.com	2481	24681	200
22 	13568436656	192.168.100.19			1116	954	200

(2) 期望输出数据
手机号136、137、138、139开头都分别放到一个独立的4个文件中，其他开头的放到一个文件中。

2.需求分析
增加一个ProvincePartitioner分区

136    分区0
137    分区1
138    分区2
139    分区3
其它    分区4

设置Driver类

// 指定自定义数据分区
job.setPartitionerClass(ProvincePartitioner.class)
// 同时指定相应数量的reduce task
job.setNumReduceTasks(5)

3.增加一个分区类

package com.test.mapreduce.flowsum;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;

public class ProvincePartitioner extends Partitioner<Text, FlowBean> {

	@Override
	public int getPartition(Text key, FlowBean value, int numPartitions) {

		// 1 获取电话号码的前三位
		String preNum = key.toString().substring(0, 3);
		
		int partition = 4;
		
		// 2 判断是哪个省
		if ("136".equals(preNum)) {
			partition = 0;
		}else if ("137".equals(preNum)) {
			partition = 1;
		}else if ("138".equals(preNum)) {
			partition = 2;
		}else if ("139".equals(preNum)) {
			partition = 3;
		}

		return partition;
	}
}

4.在驱动函数中增加自定义数据分区设置和ReduceTask设置

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

public class FlowsumDriver {

	public static void main(String[] args) throws IllegalArgumentException, IOException, ClassNotFoundException, InterruptedException {

		// 输入输出路径需要根据自己电脑上实际的输入输出路径设置
		args = new String[]{"e:/output1","e:/output2"};

		// 1 获取配置信息，或者job对象实例
		Configuration configuration = new Configuration();
		Job job = Job.getInstance(configuration);

		// 2 指定本程序的jar包所在的本地路径
		job.setJarByClass(FlowsumDriver.class);

		// 3 指定本业务job要使用的mapper/Reducer业务类
		job.setMapperClass(FlowCountMapper.class);
		job.setReducerClass(FlowCountReducer.class);

		// 4 指定mapper输出数据的kv类型
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(FlowBean.class);

		// 5 指定最终输出的数据的kv类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(FlowBean.class);

		// 8 指定自定义数据分区
		job.setPartitionerClass(ProvincePartitioner.class);

		// 9 同时指定相应数量的reduce task
		job.setNumReduceTasks(5);
		
		// 6 指定job的输入原始文件所在目录
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		// 7 将job中配置的相关参数，以及job所用的java类所在的jar包， 提交给yarn去运行
		boolean result = job.waitForCompletion(true);
		System.exit(result ? 0 : 1);
	}
}

3.7.4 WritableComparable排序

1.排序概述
排序是MapReduce框架中最重要的操作之一。

MapTask和ReduceTask均会对数据按照key进行排序。该操作属于Hadoop的默认行为。任何应用程序中的数据均会被排序，而不管逻辑上是否需要。

默认排序是按照字典顺序排序，且实现该排序的方法是快速排序。

默认ReduceTask数量为1。

2.排序的时间
对于MapTask，它会将处理的结果暂时放到环形缓冲区中，当环形缓冲区使用率达到一定阈值后，再对缓冲区中的数据进行一次快速排序，并将这些有序数据溢写到磁盘上，而当数据处理完毕后，它会对磁盘上所有文件进行归并排序。

对于ReduceTask，它从每个MapTask上远程拷贝相应的数据文件，如果文件大小超过一定阈值，则溢写磁盘上，否则存储在内存中。如果磁盘上文件数目达到一定阈值，则进行一次归并排序以生成一个更大文件；如果内存中文件大小或者数目超过一定阈值，则进行一次合并后将数据溢写到磁盘上。当所有数据拷贝完毕后，ReduceTask统一对内存和磁盘上的所有数据进行一次归并排序。

默认ReduceTask数量为1。

3.排序的分类

• 部分排序，MapReduce根据输入记录的键对数据集排序，保证输出的每个文件内部有序；
• 全排序，最终结果只有一个文件，且文件内部有序。实现该方式的操作为只设置一个ReduceTask。但该方法在处理大型文件时效率极低，因为一台机器处理所有文件，完全丧失了MapReduce所提供的并行架构。
• 辅助排序（GroupingComparator），在Reduce端对key进行排序，在接收的key为bean对象时，想让一个或几个字段相同（全部字段比较不同）的key进入到同一个reduce方法时，可以采用分组排序。
• 二次排序，在自定义排序过程中，如果compareTo中的判断条件为两个，即为二次排序。

4.自定义排序WritableComparable
bean对象做为key传输，需要实现WritableComparable接口重写compareTo方法，就可以实现排序。

@Override
public int compareTo(FlowBean o) {

	int result;
		
	// 按照总流量大小，倒序排列
	if (sumFlow > bean.getSumFlow()) {
		    result = -1;
	}else if (sumFlow < bean.getSumFlow()) {
		    result = 1;
	}else {
		    result = 0;
	}

	return result;
}

3.7.5 WritableComparable排序案例（全排序）

1.需求
根据案例2.3产生的结果再次对总流量进行排序。

(1) 输入数据

原始数据                     第一次处理后的数据
phone_data.txt            part-r-0000

(2) 期望输出数据

13509468723	7335	110349	117684
13736230513	2481	24681	27162
13956435636	132		1512	1644
13846544121	264		0		264
......

2.需求分析

3.代码实现
(1) FlowBean对象在需求1基础上增加了比较功能

package com.test.mapreduce.sort;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.WritableComparable;

public class FlowBean implements WritableComparable<FlowBean> {

	private long upFlow;
	private long downFlow;
	private long sumFlow;

	// 反序列化时，需要反射调用空参构造函数，所以必须有
	public FlowBean() {
		super();
	}

	public FlowBean(long upFlow, long downFlow) {
		super();
		this.upFlow = upFlow;
		this.downFlow = downFlow;
		this.sumFlow = upFlow + downFlow;
	}

	public void set(long upFlow, long downFlow) {
		this.upFlow = upFlow;
		this.downFlow = downFlow;
		this.sumFlow = upFlow + downFlow;
	}

	public long getSumFlow() {
		return sumFlow;
	}

	public void setSumFlow(long sumFlow) {
		this.sumFlow = sumFlow;
    }	

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

	/**
	 * 序列化方法
	 * @param out
	 * @throws IOException
	 */
	@Override
	public void write(DataOutput out) throws IOException {
		out.writeLong(upFlow);
		out.writeLong(downFlow);
		out.writeLong(sumFlow);
	}

	/**
	 * 反序列化方法 注意反序列化的顺序和序列化的顺序完全一致
	 * @param in
	 * @throws IOException
	 */
	@Override
	public void readFields(DataInput in) throws IOException {
		upFlow = in.readLong();
		downFlow = in.readLong();
		sumFlow = in.readLong();
	}

	@Override
	public String toString() {
		return upFlow + "\t" + downFlow + "\t" + sumFlow;
	}

	@Override
	public int compareTo(FlowBean bean) {
		
		int result;
		
		// 按照总流量大小，倒序排列
		if (sumFlow > bean.getSumFlow()) {
			    result = -1;
		}else if (sumFlow < bean.getSumFlow()) {
			    result = 1;
		}else {
			    result = 0;
		}
		return result;
	}
}

(2) 编写Mapper类

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

public class FlowCountSortMapper extends Mapper<LongWritable, Text, FlowBean, Text>{

	FlowBean bean = new FlowBean();
	Text v = new Text();

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

		// 1 获取一行
		String line = value.toString();
		
		// 2 截取
		String[] fields = line.split("\t");
		
		// 3 封装对象
		String phoneNbr = fields[0];
		long upFlow = Long.parseLong(fields[1]);
		long downFlow = Long.parseLong(fields[2]);
		
		bean.set(upFlow, downFlow);
		v.set(phoneNbr);
		
		// 4 输出
		context.write(bean, v);
	}
}

(3) 编写Reducer类

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

public class FlowCountSortReducer extends Reducer<FlowBean, Text, Text, FlowBean>{

	@Override
	protected void reduce(FlowBean key, Iterable<Text> values, Context context)	throws IOException, InterruptedException {
		
		// 循环输出，避免总流量相同情况
		for (Text text : values) {
			context.write(text, key);
		}
	}
}

(4) 编写Driver类

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

public class FlowCountSortDriver {

	public static void main(String[] args) throws ClassNotFoundException, IOException, InterruptedException {

		// 输入输出路径需要根据自己电脑上实际的输入输出路径设置
		args = new String[]{"e:/output1","e:/output2"};

		// 1 获取配置信息，或者job对象实例
		Configuration configuration = new Configuration();
		Job job = Job.getInstance(configuration);

		// 2 指定本程序的jar包所在的本地路径
		job.setJarByClass(FlowCountSortDriver.class);

		// 3 指定本业务job要使用的mapper/Reducer业务类
		job.setMapperClass(FlowCountSortMapper.class);
		job.setReducerClass(FlowCountSortReducer.class);

		// 4 指定mapper输出数据的kv类型
		job.setMapOutputKeyClass(FlowBean.class);
		job.setMapOutputValueClass(Text.class);

		// 5 指定最终输出的数据的kv类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(FlowBean.class);

		// 6 指定job的输入原始文件所在目录
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		
		// 7 将job中配置的相关参数，以及job所用的java类所在的jar包， 提交给yarn去运行
		boolean result = job.waitForCompletion(true);
		System.exit(result ? 0 : 1);
	}
}

3.7.6 WritableComparable排序案例（区内排序）

1.需求
要求每个省份手机号输出的文件中按照总流量内部排序。
2.需求分析
基于前一个需求，增加自定义分区类，分区按照省份手机号设置。

3.案例
(1) 增加自定义分区类

package com.test.mapreduce.sort;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;

public class ProvincePartitioner extends Partitioner<FlowBean, Text> {

	@Override
	public int getPartition(FlowBean key, Text value, int numPartitions) {
		
		// 1 获取手机号码前三位
		String preNum = value.toString().substring(0, 3);
		
		int partition = 4;
		
		// 2 根据手机号归属地设置分区
		if ("136".equals(preNum)) {
			    partition = 0;
		}else if ("137".equals(preNum)) {
			partition = 1;
		}else if ("138".equals(preNum)) {
			    partition = 2;
		}else if ("139".equals(preNum)) {
			partition = 3;
		}

		return partition;
	}
}

(2) 在驱动类中添加分区类

// 加载自定义分区类
job.setPartitionerClass(ProvincePartitioner.class);

// 设置Reducetask个数
job.setNumReduceTasks(5);

3.7.7 Combiner合并

Combiner合并是Hadoop框架优化的一种手段，因为Combiner合并减少了数据的IO传输。

(1) combiner是MR程序中Mapper和Reducer之外的一种组件。
(2) combiner组件的父类就是Reducer。
(3) combiner和reducer的区别在于运行的位置
?? Combiner是在每一个MapTask所在的节点运行;
?? Reducer是接收全局所有Mapper的输出结果；

(4) combiner的意义就是对每一个MapTask的输出进行局部汇总，以减小网络传输量。
(5) combiner能够应用的前提是不能影响最终的业务逻辑，而且，combiner的输出kv应该跟reducer的输入kv类型要对应起来。

自定义Combiner实现步骤
自定义一个Combiner继承Reducer，重写Reduce方法

public class WordcountCombiner extends Reducer<Text, IntWritable, Text,IntWritable>{

	@Override
	protected void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException {

        // 1 汇总操作
		int count = 0;
		for(IntWritable v :values){
			count += v.get();
		}

        // 2 写出
		context.write(key, new IntWritable(count));
	}
}

在Job驱动类中设置

job.setCombinerClass(WordcountCombiner.class); 

3.7.8 Combiner合并案例

1.需求
统计过程中对每一个MapTask的输出进行局部汇总，以减小网络传输量即采用Combiner功能。

(1) 数据输入

banzhang ni hao
xihuan hadoop banzhang
banzhang ni hao
xihuan hadoop banzhang

(2) 期望输出数据
期望：Combine输入数据多，输出时经过合并，输出数据降低。

2.需求分析

3.方案一
(1) 增加一个WordcountCombiner类继承Reducer

package com.test.mr.combiner;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class WordcountCombiner extends Reducer<Text, IntWritable, Text, IntWritable>{

    IntWritable v = new IntWritable();

	@Override
	protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {

        // 1 汇总
		int sum = 0;

		for(IntWritable value :values){
			sum += value.get();
		}

		v.set(sum);

		// 2 写出
		context.write(key, v);
	}
}

(2) 在WordcountDriver驱动类中指定Combiner

// 指定需要使用combiner，以及用哪个类作为combiner的逻辑
job.setCombinerClass(WordcountCombiner.class);

4.方案二
将WordcountReducer作为Combiner在WordcountDriver驱动类中指定

// 指定需要使用Combiner，以及用哪个类作为Combiner的逻辑
job.setCombinerClass(WordcountReducer.class);

运行程序，如图所示

3.7.9 GroupingComparator分组（辅助排序）

对Reduce阶段的数据根据某一个或几个字段进行分组。

分组排序步骤
(1) 自定义类继承WritableComparator
(2) 重写compare()方法

@Override
public int compare(WritableComparable a, WritableComparable b) {
	// 比较的业务逻辑
	return result;
}

(3) 创建一个构造将比较对象的类传给父类

protected OrderGroupingComparator() {
		super(OrderBean.class, true);
}

3.7.10 GroupingComparator分组案例

1.需求
有如下订单数据，需要求出每一个订单中最贵的商品。

(1) 期望输出数据

1	222.8
2	722.4
3	232.8

2.需求分析
(1) 利用“订单id和成交金额”作为key，可以将Map阶段读取到的所有订单数据按照id升序排序，如果id相同再按照金额降序排序，发送到Reduce。
(2) 在Reduce端利用GroupingComparator将订单id相同的kv聚合成组，然后取第一个即是该订单中最贵商品，如图所示。

3.代码实现
(1) 定义订单信息OrderBean类

package com.test.mapreduce.order;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.WritableComparable;

public class OrderBean implements WritableComparable<OrderBean> {

	private int order_id; // 订单id号
	private double price; // 价格

	public OrderBean() {
		super();
	}

	public OrderBean(int order_id, double price) {
		super();
		this.order_id = order_id;
		this.price = price;
	}

	@Override
	public void write(DataOutput out) throws IOException {
		out.writeInt(order_id);
        out.writeDouble(price);
	}

	@Override
	public void readFields(DataInput in) throws IOException {
		order_id = in.readInt();
		price = in.readDouble();
	}

	@Override
	public String toString() {
		return order_id + "\t" + price;
	}

	public int getOrder_id() {
		return order_id;
	}

	public void setOrder_id(int order_id) {
		this.order_id = order_id;
	}

	public double getPrice() {
		return price;
	}

	public void setPrice(double price) {
		this.price = price;
	}

	// 二次排序
	@Override
	public int compareTo(OrderBean o) {

		int result;

		if (order_id > o.getOrder_id()) {
		    result = 1;
		} else if (order_id < o.getOrder_id()) {
			result = -1;
		} else {
			// 价格倒序排序
			result = price > o.getPrice() ? -1 : 1;
		}

		return result;
	}
}

(2) 编写OrderSortMapper类

package com.test.mapreduce.order;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class OrderMapper extends Mapper<LongWritable, Text, OrderBean, NullWritable> {

	OrderBean k = new OrderBean();
	
	@Override
	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		
		// 1 获取一行
		String line = value.toString();
		
		// 2 截取
		String[] fields = line.split("\t");
		
		// 3 封装对象
		k.setOrder_id(Integer.parseInt(fields[0]));
		k.setPrice(Double.parseDouble(fields[2]));
		
		// 4 写出
		context.write(k, NullWritable.get());
	}
}

(3) 编写OrderSortGroupingComparator类

package com.test.mapreduce.order;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;

public class OrderGroupingComparator extends WritableComparator {

	protected OrderGroupingComparator() {
		    super(OrderBean.class, true);
	}

	@Override
	public int compare(WritableComparable a, WritableComparable b) {

		OrderBean aBean = (OrderBean) a;
		OrderBean bBean = (OrderBean) b;

		int result;
		if (aBean.getOrder_id() > bBean.getOrder_id()) {
			    result = 1;
		} else if (aBean.getOrder_id() < bBean.getOrder_id()) {
			    result = -1;
		} else {
			    result = 0;
		}

		return result;
	}
}

(4) 编写OrderSortReducer类

package com.test.mapreduce.order;
import java.io.IOException;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;

public class OrderReducer extends Reducer<OrderBean, NullWritable, OrderBean, NullWritable> {

	@Override
	protected void reduce(OrderBean key, Iterable<NullWritable> values, Context context)		throws IOException, InterruptedException {
		
		context.write(key, NullWritable.get());
	}
}

(5) 编写OrderSortDriver类

package com.test.mapreduce.order;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class OrderDriver {

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

        // 输入输出路径需要根据自己电脑上实际的输入输出路径设置
		args  = new String[]{"e:/input/inputorder" , "e:/output1"};

		// 1 获取配置信息
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);

		// 2 设置jar包加载路径
		job.setJarByClass(OrderDriver.class);

		// 3 加载map/reduce类
		job.setMapperClass(OrderMapper.class);
		job.setReducerClass(OrderReducer.class);

		// 4 设置map输出数据key和value类型
		job.setMapOutputKeyClass(OrderBean.class);
		job.setMapOutputValueClass(NullWritable.class);

		// 5 设置最终输出数据的key和value类型
		job.setOutputKeyClass(OrderBean.class);
		job.setOutputValueClass(NullWritable.class);

		// 6 设置输入数据和输出数据路径
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		// 8 设置reduce端的分组
        job.setGroupingComparatorClass(OrderGroupingComparator.class);

		// 7 提交
		boolean result = job.waitForCompletion(true);
		System.exit(result ? 0 : 1);
	}
}

3.8 MapReduce之Mapper类,Reducer类中的函数

Mapper类中有setup()，map()，cleanup()和run()四个方法。其中setup()一般是用来进行一些map()前的准备工作，map()则一般承担主要的处理工作，cleanup()则是收尾工作如关闭文件或者执行map()后的K-V分发等。run()方法提供了setup->map->cleanup的执行模板。

在MapReduce中，Mapper从一个输入分片中读取数据，然后经过Shuffle and Sort阶段，分发数据给Reducer，在Map端和Reduce端可能使用设置的Combiner进行合并，这在Reduce前进行。Partitioner控制每个K-V对应该被分发到哪个reducer[Job可能有多个reducer]，Hadoop默认使用HashPartitioner，HashPartitioner使用key的hashCode对reducer的数量取模得来。

Mapper类4个函数

protected void setup(Mapper.Context context) throws IOException,InterruptedException //Called once at the beginning of the task

protected void cleanup(Mapper.Context context)throws IOException,InterruptedException //Called once at the end of the task. 

protected void map(KEYIN key, VALUEIN value Mapper.Context context)throws IOException,InterruptedException
{
   context.write((KEYOUT) key,(VALUEOUT) value);
}
//Called once for each key/value pair in the input split. Most applications should override this, but the default is the identity function. 

public void run(Mapper.Context context)throws IOException,InterruptedException
{
    setup(context);
    while(context.nextKeyValue())
    {
      map(context.getCurrentKey(),context.getCurrentValue(),context)
    }
    cleanup(context);
}
//Expert users can override this method for more complete control over the execution of the Mapper. 

执行顺序：setup —> map/run ----> cleanup

Reducer类中4个函数

protected void setup(Mapper.Context context) throws IOException,InterruptedException //Called once at the beginning of the task

protected void cleanup(Mapper.Context context)throws IOException,InterruptedException //Called once at the end of the task. 

protected void reduce(KEYIN key, VALUEIN value Reducer.Context context)throws IOException,InterruptedException
{

    for(VALUEIN value: values) {
      context.write((KEYOUT) key, (VALUEOUT) value);
    }
}
//This method is called once for each key. Most applications will define their reduce class by overriding this method. The default implementation is an identity function. 

public void run(Reducer.Context context)throws IOException,InterruptedException
{
    setup(context);
    while (context.nextKey()) {
      
    reduce(context.getCurrentKey(), context.getValues(), context);
    // If a back up store is used, reset it
      
    ((ReduceContext.ValueIterator(context.getValues().iterator())).resetBackupStore();
    }
    
cleanup(context);
  }
}
//Advanced application writers can use the run(org.apache.hadoop.mapreduce.Reducer.Context) method to control how the reduce task works

执行顺序：setup —> map/run ----> cleanup

3.9 OutputFormat数据输出

3.9.1OutputFormat接口实现类

OutputFormat是MapReduce输出的基类，所有实现MapReduce输出格式化的类都实现了 OutputFormat接口。

文本输出TextOutputFormat
默认的输出格式是TextOutputFormat，它把每条记录写为文本行。它的键和值可以是任意类型，因为TextOutputFormat调用toString()方法把它们转换为字符串

SequenceFileOutputFormat
将SequenceFileOutputFormat输出作为后续 MapReduce任务的输入，这便是一种好的输出格式，因为它的格式紧凑，很容易被压缩。

自定义OutputFormat
根据用户需求，自定义实现输出。

3.9.2 自定义OutputFormat

为了控制最终文件的输出路径和输出格式，可以自定义OutputFormat。

例如，要在一个MapReduce程序中根据数据的不同，输出两类结果到不同目录，这类灵活的输出需求可以通过自定义OutputFormat来实现。

自定义OutputFormat步骤
(1) 自定义一个类继承FileOutputFormat。
(2) 改写RecordWriter，具体改写输出数据的方法write()。

3.9.3 自定义OutputFormat案例

1.需求
过滤输入的log日志，包含test的网站输出到e:/test.log，不包含test的网站输出到e:/other.log。
(1) 输入数据log.txt

http://www.baidu.com
http://www.google.com
http://cn.bing.com
http://www.test.com
http://www.sohu.com
http://www.sina.com
http://www.sin2a.com
http://www.sin2desa.com
http://www.sindsafa.com

(2) 期望输出数据
test.log

http://www.test.com

other.log

http://cn.bing.com
http://www.baidu.com
http://www.google.com
http://www.sin2a.com
http://www.sin2desa.com
http://www.sina.com
http://www.sindsafa.com
http://www.sohu.com

2.案例实操
(1) 编写FilterMapper类

package com.test.mapreduce.outputformat;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class FilterMapper extends Mapper<LongWritable, Text, Text, NullWritable>{
	
	@Override
	protected void map(LongWritable key, Text value, Context context)	throws IOException, InterruptedException {

		// 写出
		context.write(value, NullWritable.get());
	}
}

(2) 编写FilterReducer类

package com.test.mapreduce.outputformat;
import java.io.IOException;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class FilterReducer extends Reducer<Text, NullWritable, Text, NullWritable> {

Text k = new Text();

	@Override
	protected void reduce(Text key, Iterable<NullWritable> values, Context context)		throws IOException, InterruptedException {

       // 1 获取一行
		String line = key.toString();

       // 2 拼接
		line = line + "\r\n";

       // 3 设置key
       k.set(line);

       // 4 输出
		context.write(k, NullWritable.get());
	}
}

(3) 自定义一个OutputFormat类

package com.test.mapreduce.outputformat;
import java.io.IOException;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class FilterOutputFormat extends FileOutputFormat<Text, NullWritable>{

	@Override
	public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext job)			throws IOException, InterruptedException {

		// 创建一个RecordWriter
		return new FilterRecordWriter(job);
	}
}

(4) 编写RecordWriter类

package com.test.mapreduce.outputformat;
import java.io.IOException;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

public class FilterRecordWriter extends RecordWriter<Text, NullWritable> {

	FSDataOutputStream testOut = null;
	FSDataOutputStream otherOut = null;

	public FilterRecordWriter(TaskAttemptContext job) {

		// 1 获取文件系统
		FileSystem fs;

		try {
			fs = FileSystem.get(job.getConfiguration());

			// 2 创建输出文件路径
			Path testPath = new Path("e:/test.log");
			Path otherPath = new Path("e:/other.log");

			// 3 创建输出流
			testOut = fs.create(testPath);
			otherOut = fs.create(otherPath);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	@Override
	public void write(Text key, NullWritable value) throws IOException, InterruptedException {

		// 判断是否包含“test”输出到不同文件
		if (key.toString().contains("test")) {
			    testOut.write(key.toString().getBytes());
		} else {
			    otherOut.write(key.toString().getBytes());
		}
	}

	@Override
	public void close(TaskAttemptContext context) throws IOException, InterruptedException {

		    // 关闭资源
        IOUtils.closeStream(testOut);
		IOUtils.closeStream(otherOut);	}
}

(5) 编写FilterDriver类

```java
package com.test.mapreduce.outputformat;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
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 FilterDriver {

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

// 输入输出路径需要根据自己电脑上实际的输入输出路径设置
args = new String[] { "e:/input/inputoutputformat", "e:/output2" };

		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);

		job.setJarByClass(FilterDriver.class);
		job.setMapperClass(FilterMapper.class);
		job.setReducerClass(FilterReducer.class);

		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(NullWritable.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);

		// 要将自定义的输出格式组件设置到job中
		job.setOutputFormatClass(FilterOutputFormat.class);

		FileInputFormat.setInputPaths(job, new Path(args[0]));

		// 虽然我们自定义了outputformat，但是因为我们的outputformat继承自fileoutputformat
		// 而fileoutputformat要输出一个_SUCCESS文件，所以，在这还得指定一个输出目录
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		boolean result = job.waitForCompletion(true);
		System.exit(result ? 0 : 1);
	}
}

3.10 Join多种应用

3.10.1 Reduce Join

Map 端主要工作：对不同表或文件的 key/value，打标签以区别不同来源的记录。然后用连接字段作为 key，其余部分和新加的标志作为 value，最后进行输出。

Reduce 端主要工作：在 Reduce 端以连接字段作为 key 的分组已经完成，只需要在每一个分组当中将那些来源于不同文件的记录（在 Map 阶段已经打标志）分开，最后进行合并。

3.10.2 Reduce Join案例

1.需求
订单数据表t_order

id	pid	amount
1001	01	1
1002	02	2
1003	03	3
1004	01	4
1005	02	5
1006	03	6

商品信息表t_product

pid	pname
01	小米
02	华为
03	格力

将商品信息表中数据根据商品pid合并到订单数据表中。
表最终数据形式

id	pname	amount
1001	小米	1
1004	小米	4
1002	华为	2
1005	华为	5
1003	格力	3
1006	格力	6

2.需求分析
通过将关联条件作为Map输出的key，将两表满足Join条件的数据并携带数据来源的文件信息，发往同一个ReduceTask，在Reduce中进行数据的串联，如图所示。

3.代码实现
(1) 创建商品和订单合并后的Bean类

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

public class TableBean implements Writable {

	private String order_id; // 订单id
	private String p_id;      // 产品id
	private int amount;       // 产品数量
	private String pname;     // 产品名称
	private String flag;      // 表的标记

	public TableBean() {
		    super();
	}

	public TableBean(String order_id, String p_id, int amount, String pname, String flag) {

		super();

		this.order_id = order_id;
		this.p_id = p_id;
		this.amount = amount;
		this.pname = pname;
		this.flag = flag;
	}

	public String getFlag() {
		return flag;
	}

	public void setFlag(String flag) {
		this.flag = flag;
	}

	public String getOrder_id() {
		return order_id;
	}

	public void setOrder_id(String order_id) {
		this.order_id = order_id;
	}

	public String getP_id() {
		return p_id;
	}

	public void setP_id(String p_id) {
		this.p_id = p_id;
	}

	public int getAmount() {
		return amount;
	}

	public void setAmount(int amount) {
		this.amount = amount;
	}

	public String getPname() {
		return pname;
	}

	public void setPname(String pname) {
		this.pname = pname;
	}

	@Override
	public void write(DataOutput out) throws IOException {
		out.writeUTF(order_id);
		out.writeUTF(p_id);
		out.writeInt(amount);
		out.writeUTF(pname);
		out.writeUTF(flag);
	}

	@Override
	public void readFields(DataInput in) throws IOException {
		this.order_id = in.readUTF();
		this.p_id = in.readUTF();
		this.amount = in.readInt();
		this.pname = in.readUTF();
		this.flag = in.readUTF();
	}

	@Override
	public String toString() {
		return order_id + "\t" + pname + "\t" + amount + "\t" ;
	}
}

(2) 编写TableMapper类

package com.test.mapreduce.table;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

public class TableMapper extends Mapper<LongWritable, Text, Text, TableBean>{

String name;
	TableBean bean = new TableBean();
	Text k = new Text();
	
	@Override
	protected void setup(Context context) throws IOException, InterruptedException {

		// 1 获取输入文件切片
		FileSplit split = (FileSplit) context.getInputSplit();

		// 2 获取输入文件名称
		name = split.getPath().getName();
	}

	@Override
	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		
		// 1 获取输入数据
		String line = value.toString();
		
		// 2 不同文件分别处理
		if (name.startsWith("order")) {// 订单表处理

			// 2.1 切割
			String[] fields = line.split("\t");
			
			// 2.2 封装bean对象
			bean.setOrder_id(fields[0]);
			bean.setP_id(fields[1]);
			bean.setAmount(Integer.parseInt(fields[2]));
			bean.setPname("");
			bean.setFlag("order");
			
			k.set(fields[1]);
		}else {// 产品表处理

			// 2.3 切割
			String[] fields = line.split("\t");
			
			// 2.4 封装bean对象
			bean.setP_id(fields[0]);
			bean.setPname(fields[1]);
			bean.setFlag("pd");
			bean.setAmount(0);
			bean.setOrder_id("");
			
			k.set(fields[0]);
		}

		// 3 写出
		context.write(k, bean);
	}
}

(3) 编写TableReducer类

package com.test.mapreduce.table;
import java.io.IOException;
import java.util.ArrayList;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class TableReducer extends Reducer<Text, TableBean, TableBean, NullWritable> {

	@Override
	protected void reduce(Text key, Iterable<TableBean> values, Context context)	throws IOException, InterruptedException {

		// 1准备存储订单的集合
		ArrayList<TableBean> orderBeans = new ArrayList<>();
		
// 2 准备bean对象
		TableBean pdBean = new TableBean();

		for (TableBean bean : values) {

			if ("order".equals(bean.getFlag())) {// 订单表

				// 拷贝传递过来的每条订单数据到集合中
				TableBean orderBean = new TableBean();

				try {
					BeanUtils.copyProperties(orderBean, bean);
				} catch (Exception e) {
					e.printStackTrace();
				}
                orderBeans.add(orderBean);
			} else {// 产品表

				try {
					// 拷贝传递过来的产品表到内存中
					BeanUtils.copyProperties(pdBean, bean);
				} catch (Exception e) {
					e.printStackTrace();
			    }
			}
		}

		// 3 表的拼接
		for(TableBean bean:orderBeans){

			bean.setPname (pdBean.getPname());
			
			// 4 数据写出去
			context.write(bean, NullWritable.get());
		}
	}
}

(4) 编写TableDriver类

package com.test.mapreduce.table;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
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 TableDriver {

	public static void main(String[] args) throws Exception {
		
        // 0 根据自己电脑路径重新配置
        args = new String[]{"e:/input/inputtable","e:/output1"};

        // 1 获取配置信息，或者job对象实例
		Configuration configuration = new Configuration();
		Job job = Job.getInstance(configuration);

		// 2 指定本程序的jar包所在的本地路径
		job.setJarByClass(TableDriver.class);

		// 3 指定本业务job要使用的Mapper/Reducer业务类
		job.setMapperClass(TableMapper.class);
		job.setReducerClass(TableReducer.class);

		// 4 指定Mapper输出数据的kv类型
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(TableBean.class);

		// 5 指定最终输出的数据的kv类型
		job.setOutputKeyClass(TableBean.class);
		job.setOutputValueClass(NullWritable.class);

		// 6 指定job的输入原始文件所在目录
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		// 7 将job中配置的相关参数，以及job所用的java类所在的jar包， 提交给yarn去运行
		boolean result = job.waitForCompletion(true);
		System.exit(result ? 0 : 1);
	}
}