如何查看hadoop mapreduce 性能

Posted 2023-04-15

1) 优化map,reduce任务运行的数量
症状：
每个 map 或 reduce 任务都在30-40秒内结束。一个大job没有使用上所有集群中的可用槽位。在大部分mapper和reducer都订好运行计划后，1到2个仍在pending状态直到最后才单独运行。 诊断:优化map和reduce的任务是非常重要但是经常被忽视，这里介绍几个我常用的相关设置方法：

如果每个任务只执行30-40秒就结束，请减少总的task数量。Task的基本设置和计划本身会消耗几秒钟的时间。所以如果Task执行非常快的话，时间就都浪费在准备Task上了。也可以开启JVM的reuse功能来减少建立task的基本开销。如果job要处理超过1TB的数据，可以考虑增加输入数据的块Block的大小从256MB到512MB。这样也会减小需要运行的Task数。可以通过如下命令改变数据块大小：hadoop distcp -Ddfs.block.size=$[256*1024*1024] /path/to/inputdata /path/to/inputdata-with-largeblocks.执行完该命令就可以清除原来的文件了。在保证每个任务执行都超过30-40秒后，可以增加mapper task为mapper slot(可以执行mapper 机器)的整数倍，如果你有100个可以运行Map任务的节点，尽量不要运行101个Map Task，第101个Map task 会在第一批100个Map任务执行完之后才执行，这点主要针对的是小型集群和小型任务。
不要计划执行太多的Reduce任务，对于大多数任务，我们建议Reduce任务数要等于或小于集群中可运行Reduce任务的节点数。
性能测试:
我使用一个参数-Dmapred.max.split.size=$[16*1024*1024] 来展示设置了过多任务的wordcount程序。这样会产生2640个而不是默认的360个任务来执行该程序。当以这种配置运行时单个的任务平均只用9秒，在JobTracker的监控页面上可以看到正在map任务数在0到24之间波动，整个Job花了17分52秒，是原来配置的2倍。

2) 在集群上使用 LZO 压缩插件

症状：
*应用于中间数据LZO压缩始终是个好方法。
*MapReduce 任务输出文件尺寸很大。
*在任务运行时Slave节点上top和iostat中显示高iowait。

诊断：
几乎任何产生大量map输出的MapReduce任务都能从LZO压缩算法受益。虽然LZO增加了一些CPU的负载，但是shuffle阶段减少的大量磁盘IO操作会把时间完全节省回来。
当job要处理大量数据时，LZO压缩也可以增加输出方面的的性能。在默认的3份复制配置下，每1GB压缩省下的空间都相当于节省了3GB的IO写操作。
要开启LZO压缩，请见另一篇文章，

记得要把mapred.compress.map.output设为true。

性能对比:
禁用LZO只在测试中轻微延长了运行时间。但是文件写出量计数FILE_BYTESwww.hbbz08.com_WRITTEN从3.5G增长到9.2G，显示出62%的IO优化效果，在一个job独自运行的环境下，IO并不是瓶颈，所以时间缩短并不明显。当在高任务并发的集群上运行时，60%的IO减少会带来明显的速度提升。

3) 正确配置Hadoop集群
症状：
*当所有的MapReduce任务栏位都在运行任务时，用top命令观察到slave节点仍然相对的空闲。
*用top观察到内核进程RAID(mdX_raid*)或pdflush占用大量CPU
*Linux平均负载经常高于系统CPU数x2
*执行任务时，Linux平均负载低于系统CPU数
*节点上超过几MB的SWAP使用量
诊断： 参考技术A

可以只用一行代码来运行MapReduce作业：JobClient.runJon(conf)，Job作业运行时参与的四个实体：

     1.JobClient 写代码，配置作业，提交作业。

     2.JobTracker:初始化作业，分配作业，协调作业运行。这是一个java程序，主类是JobTracker。

     3.TaskTracker：运行作业划分后的任务，即分配数据分配上执行Map或Reduce任务。

     4.HDFS：保存作业数据、配置信息等，保存作业结果。

Map/Reduce 作业总体执行流程：

     代码编写 ----> 作业配置  ---->  作业提交 ----> Map任务分配和执行 ----> 处理中间结果 ---->  Reduce任务分配与执行 ---->  输出结果

而对于每个作业的执行，又包含：

     输入准备 ----> 任务执行 ----> 输出结果

作业提交JobClient：

     JobClient的runJob方法产生一个Jobclient实例并调用其submitJob方法，然后runJob开始循环吗，并在循环中调用getTaskCompetionEvents方法，获得TaskCompletionEvent实例，每秒轮询作业进度（后面有介绍进度和状态更新），把进度写到控制台，作业完成后显示作业计数器，若失败，则把错误记录到控制台。

     submitJob方法作业提交的过程：

     1.向JobTracker请求一个新的JobId。

     2.检查作业相关路径，如果路径不正确就会返回错误。

     3.计算作业输入分片及其划分信息。

     4.将作业运行需要的资源（jar文件、配置文件等）复制到Shared HDFS，并

复制多个副本（参数控制，默认值为10）供tasktracker访问，也会将计算的分片复制到HDFS。

     5.调用JobTracker对象的submitJob()方法来真正提交作业，告诉JobTracker作业准备执行。

作业的初始化JobTracker：

     JobTracker收到submitJob方法调用后，会把调用放入到一个内部队列，由作业调度器（Job scheduler）进行调度并对其初始化。Job初始化即创建一个作业对象。

     当作业被调度后，JobTracker会创建一个代表这个作业的JobInProgress对象，并将任务和记录信息封装在这个对象中，以便跟踪任务状态和进程。

     初始化过程就是JobInProgress对象的initTasks方法进行初始化的。

     初始化步骤：

          1.从HDFS中读取作业对应的job.split信息，为后面的初始化做好准备。

          2.创建并初始化map和reduce任务。根据数据分片信息中的个数确定map task的个数，然后为每个map task生成一个TaskInProgress对象来处理数据分片，先将其放入nonRunningMapCache,以便JobTracker分配任务的时候使用。接下来根据JobConf中的mapred.reduce.tasks属性利用setNumReduceTasks()方法设置reduce task的数量，然后同map task创建方式。

          3.最后就是创建两个初始化task，进行map和reduce的初始化。

任务的分配JobTracker：

    消息传递HeartBeat： tasktracker运行一个简单循环定期发送心跳（heartbeat）给JobTracker。由心跳告知JobTracker自己是否存活，同时作为消息通道传递其它信息（请求新task）。作为心跳的一部分，tasktracker会指明自己是否已准备好运行新的任务，如果是，jobtracker会分配它一个任务。

    分配任务所属于的作业：在Jobtracker分配任务前需先确定任务所在的作业。后面会介绍到各种作业调度算法，默认是一个FIFO的作业调度。

    分配Map和Reduce任务：tasktracker有固定数量的任务槽,一个tasktracker可以同时运行多个Map和Reduce任务，但其准确的数量由tasktracker的核的数量和内存大小决定。默认调度器会先填满Map任务槽，再填Reduce任务槽。jobtracker会选择距离离分片文件最近的tasktracker，最理想情况下，任务是数据本地化（data-local）的,当然也可以是机架本地化（rack-local），如果不是本地化的，那么他们就需要从其他机架上检索数据。Reduce任务分配很简单，jobtracker会简单的从待运行的reduce任务列表中选取下一个来执行，不用考虑数据本地化。

任务的执行TaskTracker：

     TaskTracker收到新任务后，就要在本地运行任务了，运行任务的第一步就是通过localizedJob将任务本地化所需要的注入配置、数据、程序等信息进行本地化。

     1.本地化数据：从共享文件系统将job.split 、job.jar (在分布式缓存中)复制本地，将job配置信息写入job.xml。

     2.新建本地工作目录：tasktracker会加压job.jar文件到本工作目录。

     3.调用launchTaskForJob方法发布任务（其中会新建TaskRunner实例运行任务），如果是Map任务就启用MapTaskRunner，对于Reduce就是ReduceTaskRunner。

   

  在这之后，TaskRunner会启用一个新的JVM来运行每个Map/Reduce任务，防止程序原因而导致tasktracker崩溃，但不同任务间重用JVM还是可以的，后续会讲到任务JVM重用。

     对于单个Map，任务执行的简单流程是：

     1.分配任务执行参数

     2.在Child临时文件中添加map任务信息（Child是运行Map和Reduce任务的主进程）

     3.配置log文件夹，配置map任务的通信和输出参数

     4.读取input split，生成RecordReader读取数据

     5.为Map生成MapRunnable,依次从RecordReader中接收数据，并调用Map函数进行处理。

     6.最后将map函数的输出调用collect收集到MapOutputBuffer（参数控制其大小）中。

Streaming和Pipes:

     Streaming和Pipes都运行特殊的Map和Reduce任务，目的是运行用户提供的可执行程序并与之通信。

     Streaming:使用标准输入输出Streaming与进程进行通信。

     Pipes:用来监听套接字，会发送一个端口号给C++程序，两者便可建立链接。

     

进度和状态更新：

     一个作业和它的任务都有状态（status），其中包括：运行成功失败状态、Map/Reduce进度、作业计数器值、状态消息。

     状态消息与客户端的通信：

     1.对于Map任务Progress的追踪：progress是已经处理完的输入所占的比例。

     2.对于Reduce：稍复杂，reduce任务分三个阶段（每个阶段占1/3），复制、排序和Reduce处理，若reduce已执行一半的输入的话，那么任务进度便是1/3+1/3+1/6=5/6。

     3.任务计数器：任务有一组计数器，负责对任务运行各个事件进行计数。

     4.任务进度报告：如果任务报告了进度，便会设置一个标记以表明状态将被发送到tasktracker。有一个独立线程每隔三秒检查一次此标记，如果已设置，则告知tasktracker当前状态。

     5.tasktracker进度报告：tasktracker会每隔5秒（这个心跳是由集群大小决定，集群越大时间会越长）发送heartbeat到jobtracker，并且tasktracker运行的所有状态都会在调用中被发送到jobtracker。

     6.jobtracker合并各任务报告：产生一个表明所有运行作业机器所含任务状态的全局视图。

     前面提到的JobClient就是通过每秒查询JobTracker来接收最新状态，而且客户端JobClient的getJob方法可以得到一个RunningJob的实例，其包含了作业的所以状态信息。

     

作业的完成：

     当jobtracker收到作业最后一个任务已完成的通知后，便把作业状态设置成成功。JobClient查询状态时，便知道任务已成功完成，于是JobClient打印一条消息告知用户，然后从runJob方法返回。

     如果jobtracker有相应设置，也会发送一个Http作业通知给客户端，希望收到回调指令的客户端可以通过job.end.notification.url属性来进行设置。

     jobtracker情况作业的工作状态，指示tasktracker也清空作业的工作状态，如删除中间输出。

     

失败

     实际情况下，用户的代码存在软件错误进程会崩溃，机器也会产生故障，但Hadoop能很好的应对这些故障并完成作业。

     1.任务失败    

     子任务异常：如Map/Reduce任务中的用户代码抛出异常，子任务JVM进程会在退出前向父进程tasktracker发送错误报告，错误被记录用户日志。tasktracker会将此次task attempt标记为tailed，并释放这个任务槽运行另外一个任务。

     子进程JVM突然退出：可能由于JVM bug导致用户代码造成的某些特殊原因导致JVM退出，这种情况下，tasktracker会注意到进程已经退出，并将此次尝试标记为failed。

     任务挂起：一旦tasktracker注意一段时间没有收到进度更新，便会将任务标记为failed，JVM子进程将被自动杀死。任务失败间隔时间通常为10分钟，可以以作业或者集群为基础设置过期时间，参数为mapred.task.timeout。注意：如果参数值设置为0，则挂起的任务永远不会释放掉它的任务槽，随着时间的推移会降低整个集群的效率。

     任务失败尝试次数：jobtracker得知一个tasktracker失败后，它会重新调度该任务执行，当然，jobtracker会尝试避免重新调度失败过的tasktracker任务。如果一个任务尝试次数超过4次，它将不再被重试。这个值是可以设置的，对于Map任务，参数是mapred.map.max.attempts,对于reduce任务，则由mapred.reduce.max.attempts属性控制。如果次数超过限制，整个作业都会失败。当然，有时我们不希望少数几个任务失败就终止运行的整个作业，因为即使有些任务失败，作业的一些结果可能还是有用的，这种情况下，可以为作业设置在不触发作业失败情况下的允许任务失败的最大百分比，Map任务和Reduce任务可以独立控制，参数为mapred.max.map.failures.percent 和mapred.max.reduce.failures.percent。

     任务尝试中止（kill）：任务终止和任务失败不同，task attempt可以中止是因为他是一个推测副本或因为它所处的tasktracker失败，导致jobtracker将它上面的所有task attempt标记为killed。被终止的task attempt不会被计入任务运行尝试次数，因为尝试中止并不是任务的错。

     2.tasktracker失败

     tasktracker由于崩溃或者运行过慢而失败，他将停止向jobtracker发送心跳（或很少发送心跳）。jobtracker注意已停止发送心跳的tasktracker（过期时间由参数mapred.tasktracker.expiry.interval设置，单位毫秒），并将它从等待调度的tasktracker池中移除。如果是未完成的作业，jobtracker会安排次tasktracker上已经运行成功的Map任务重新运行，因为此时reduce任务已无法访问（中间输出存放在失败的tasktracker的本地文件系统上）。

     即使tasktracker没有失败，也有可能被jobtracker列入黑名单。如果tasktracker上面的失败任务数量远远高于集群的平均失败任务次数，他就会被列入黑名单，被列入黑名单的tasktracker可以通过重启从jobtracker黑名单中移除。

     3.jobtracker失败

     老版本的JobTracker失败属于单点故障，这种情况下作业注定失败。

作业调度：

     早期作业调度FIFO：按作业提交顺序先进先出。可以设置优先级，通过设置mapred.job.priority属性或者JobClient的setJobPriority()方法制定优先级（优先级别：VERY_HIGH,HIGH,NORMAL,LOW,VERY_LOW）。注意FIFO调度算法不支持抢占（preemption）,所以高优先级作业仍然会被那些已经开始的长时间运行的低优先级作业所阻塞。

     Fair Scheduler：目标是让每个用户公平地共享集群能力。当集群存在很多作业时，空闲的任务槽会以”让每个用户共享集群“的方式进行分配。默认每个用户都有自己的作业池。FairScheduler支持抢占，所以，如果一个池在特定的一段时间未得到公平地资源共享，它会终止池中得到过多的资源任务，以便把任务槽让给资源不足的池。FairScheduler是一个后续模块，使用它需要将其jar文件放在Hadoop的类路径下。可以通过参数map.red.jobtracker.taskScheduler属性配置（值为org.apache.hadoop.mapred.FairScheduler）

     Capacity Scheduler：

     集群由很多队列组成，每个队列都有一个分配能力，这一点与FairScheduler类似，只不过在每个队列内部，作业根据FIFO方式进行调度。本质上说，Capacity Scheduler允许用户或组织为每个用户模拟一个独立使用FIFO的集群。

shuffle和排序：

     MapReduce确保每个Reducer的输入都是按键排序的。系统执行排序的过程-将map输出作为输入传给reducer的过程称为shuffle。shuffle属于不断被优化和改进的代码库的一部分，从许多方面来看，shuffle是MapReduce的心脏。

     整个shuffle的流程应该是这样：

     map结果划分partition  排序sort 分割spill   合并同一划分   合并同一划分  合并结果排序 reduce处理 输出

     Map端：

     写入缓冲区：Map函数的输出，是由collector处理的，它并不是简单的将结果写到磁盘。它利用缓冲的方式写到内存，并处于效率的考虑进行预排序。每个map都有一个环形的内存缓冲区，用于任务输出，默认缓冲区大小为100MB（由参数io.sort.mb调整），一旦缓冲区内容达到阈值（默认0.8），后台进程边开始把内容写到磁盘（spill），在写磁盘过程中，map输出继续被写到缓冲区，但如果缓冲区被填满，map会阻塞知道写磁盘过程完成。写磁盘将按照轮询方式写到mapred.local.dir属性制定的作业特定子目录中。

     写出缓冲区：collect将缓冲区的内容写出时，会调用sortAndSpill函数，这个函数作用主要是创建spill文件，按照key值对数据进行排序，按照划分将数据写入文件，如果配置了combiner类，会先调用combineAndSpill函数再写文件。sortAndSpill每被调用一次，就会写一个spill文件。

     合并所有Map的spill文件：TaskTracker会在每个map任务结束后对所有map产生的spill文件进行merge，merge规则是根据分区将各个spill文件中数据同一分区中的数据合并在一起，并写入到一个已分区且排序的map输出文件中。待唯一的已分区且已排序的map输出文件写入最后一条记录后，map端的shuffle阶段就结束了。

     在写磁盘前，线程首先根据数据最终要传递到的reducer把数据划分成响应的分区（partition），在每个分区中，后台线程按键进行内排序，如果有一个combiner,它会在排序后的输出上运行。

     内存达到溢出写的阈值时，就会新建一个溢出写文件，因为map任务完成其最后一个输出记录之后，会有几个溢出写文件。在任务完成前，溢出写文件会被合并成一个已分区且已排序的输出文件。配置属性io.sort.facor控制一次最多能合并多少流，默认值是10。

     如果已经指定combiner,并且写次数至少为3（通过min.mum.spills.for.combine设置）时，则combiner就会在输出文件写到磁盘之前运行。运行combiner的意义在于使map输出更紧凑，舍得写到本地磁盘和传给reducer的数据更少。

     写磁盘时压缩：写磁盘时压缩会让写的速度更快，节约磁盘空间，并且减少传给reducer的数据量。默认情况下，输出是不压缩的，但可以通过设置mapred.compress.map.output值为true，就可以启用压缩。使用的压缩库是由mapred.map.output.compression.codec制定。

     reducer获得文件分区的工作线程：reducer通过http方式得到输出文件的分区，用于文件分区的工作线程数量由tracker.http.threads属性指定，此设置针对的是每个tasktracker，而不是每个map任务槽。默认值为40，在大型集群上此值可以根据需要而增加。

     Reduce端：

     复制阶段：reduce会定期向JobTracker获取map的输出位置，一旦拿到输出位置，reduce就会从对应的TaskTracker上复制map输出到本地（如果map输出很小，则会被复制到TaskTracker节点的内存中，否则会被让如磁盘），而不会等到所有map任务结束（当然这个也有参数控制）。

     合并阶段：从各个TaskTracker上复制的map输出文件（无论在磁盘还是内存）进行整合，并维持数据原来的顺序。

     Reduce阶段：从合并的文件中顺序拿出一条数据进行reduce函数处理，然后将结果输出到本地HDFS。

     Map的输出文件位于运行map任务的tasktracker的本地磁盘，现在，tasktracker要为分区文件运行reduce任务。每个任务完成时间可能不同，但是只要有一个任务完成，reduce任务就开始复制其输出，这就是reduce任务的复制阶段（copy phase）。reduce任务有少量复制线程，因此能够并行取得map输出。默认值是5个线程，可以通过mapred.reduce.parallel.copies属性设置。

     Reducer如何得知从哪个tasktracker获得map输出：map任务完成后会通知其父tasktracker状态已更新，tasktracker进而通知（通过heart beat）jobtracker。因此，JobTracker就知道map输出和tasktracker之间的映射关系，reducer中的一个线程定期询问jobtracker以便获知map输出位置。由于reducer有可能失败，因此tasktracker并没有在第一个reducer检索到map输出时就立即从磁盘上删除它们，相反他会等待jobtracker告示它可以删除map输出时才删除，这是作业完成后最后执行的。

     如果map输出很小，则会被直接复制到reduce tasktracker的内存缓冲区（大小由mapred.job.shuffle.input.buffer.percent控制，占堆空间的百分比），否则，map输出被复制到磁盘。一旦内存缓冲区达到阈值大小（由mapred.iob.shuffle.merge.percent）

或达到map输出阈值大小（mapred.inmem.threadhold），则合并后溢出写到磁盘中。

     随着磁盘上副本增多，后台线程会将他们合并为更大的、排好序的文件。注意：为了合并，压缩的map输出必须在内存中被解压缩。

     排序阶段：复制阶段完成后，reduce任务会进入排序阶段，更确切的说是合并阶段，这个阶段将合并map输出，维持其顺序排列。合并是循环进行的，由合并因子决定每次合并的输出文件数量。但让有可能会产生中间文件。

     reduce阶段：在最后reduce阶段，会直接把排序好的文件输入reduce函数，不会对中间文件进行再合并，最后的合并即可来自内存，也可来自磁盘。此阶段的输出会直接写到文件系统，一般为hdfs。

     细节：这里合并是并非平均合并，比如有40个文件，合并因子为10，我们并不是每趟合并10个，合并四趟。而是第一趟合并4个，后三趟合并10，在最后一趟中4个已合并的文件和余下6个未合并会直接并入reduce。

本回答被提问者采纳

Eclipse远程提交MapReduce任务到Hadoop集群

一、介绍

以前写完MapReduce任务以后总是打包上传到Hadoop集群，然后通过shell命令去启动任务，然后在各个节点上去查看Log日志文件，后来为了提高开发效率，需要找到通过Ecplise直接将MaprReduce任务直接提交到Hadoop集群中。该章节讲述用户如何从Eclipse的压缩包最终完成Eclipse提价任务给MapReduce集群。

二、详解

1、安装Eclipse，安装hadoop插件

（1）首先下载Eclipse的压缩包，然后可以从这里下载hadoop 2.7.1的ecplise插件和其他一些搭建环境中所需要的文件，然后解压ecplise，并放置到D盘中

（2）将下载的资源中的Hadoop-ecplise-plugin.jar 插件放到ecplise的插件目录中： D:\\ecplise\\plugins\\ 。然后开启ecplise。

（3）将Hadoop-2.7.1解压一份到D盘中，并配置相应的环境变量，并将%HADOOP_HOME%\\bin 文件加添加到Path环境中

（4）然后选在ecplise中配置hadoop插件：

A、Window---->show view -----> other ,在其中选中MapReduce tool

B: Window---->Perspective------>Open Perspective -----> othrer

C : Window ---->  Perferences ----> Hadoop Map/Reduce ,然后将刚刚解压的文件Hadoop文件选中

D、配置HDFS连接：该MapReduce view中创建一个新的MapReduce连接

当做完这些，我们就能在Package Exploer 中看到DFS，然后冲中可以看到HDFS上的文件：

2、进行MapReduce开发

（1）将hadoop-ecplise文件夹中的hadoopbin.zip进行解压，将会得到下列文件，并将这些文件放入到HADOOP_HOME\\bin目录下，然后将hadoop.dll文件放入到C:\\Window\\System32文件夹中

（2）从集群中下载： log4j.properties,core-site.xml,hdfs-site.xml,mapred-site.xml,yarn-site.xml 这五个文件。然后写出一个WordCount的例子，然后将这五个文件放入到src文件夹下：

（3）修改mapred-site.xml和yarn-site.xml文件

A、mapred-site.xml上添加一下几个keyvalue键值：

<property>
<name>mapred.remote.os</name>
<value>Linux</value>
</property>

<property>
<name>mapreduce.app-submission.cross-platform</name>
<value>true</value>
</property>

<property>
<name>mapreduce.application.classpath</name>
<value>/home/hadoop/hadoop/hadoop-2.7.1/etc/hadoop,
        /home/hadoop/hadoop/hadoop-2.7.1/share/hadoop/common/*,
        /home/hadoop/hadoop/hadoop-2.7.1/share/hadoop/common/lib/*,
        /home/hadoop/hadoop/hadoop-2.7.1/share/hadoop/hdfs/*,
        /home/hadoop/hadoop/hadoop-2.7.1/share/hadoop/hdfs/lib/*,
        /home/hadoop/hadoop/hadoop-2.7.1/share/hadoop/mapreduce/*,
        /home/hadoop/hadoop/hadoop-2.7.1/share/hadoop/mapreduce/lib/*,
        /home/hadoop/hadoop/hadoop-2.7.1/share/hadoop/yarn/*,
        /home/hadoop/hadoop/hadoop-2.7.1/share/hadoop/yarn/lib/*</value>
</property>

B、yarn-site.xml文件中添加一下参数：

<property>
<name>yarn.application.classpath</name>
<value>/home/hadoop/hadoop/hadoop-2.7.1/etc/hadoop,
        /home/hadoop/hadoop/hadoop-2.7.1/share/hadoop/common/*,
        /home/hadoop/hadoop/hadoop-2.7.1/share/hadoop/common/lib/*,
        /home/hadoop/hadoop/hadoop-2.7.1/share/hadoop/hdfs/*,
        /home/hadoop/hadoop/hadoop-2.7.1/share/hadoop/hdfs/lib/*,
        /home/hadoop/hadoop/hadoop-2.7.1/share/hadoop/mapreduce/*,
        /home/hadoop/hadoop/hadoop-2.7.1/share/hadoop/mapreduce/lib/*,
        /home/hadoop/hadoop/hadoop-2.7.1/share/hadoop/yarn/*,
        /home/hadoop/hadoop/hadoop-2.7.1/share/hadoop/yarn/lib/*</value>
</property>

这里需要解释一下，在Hadoop2.6之前，因为其源代码中适配了Linux操作系统中的环境变脸表示符号$，而当在window下使用这些代码是，因为两个系统之间的变量符是不一样的，所以会导致以下的错误

org.apache.hadoop.util.Shell$ExitCodeException: /bin/bash: line 0: fg: no job control 

在Hadoop2.6之前需要通过修改源代码后打jar包替换旧的Jar包文件，具体的流程请看下面这篇博客：

http://www.aboutyun.com/thread-8498-1-1.html

在这里我们通过修改mapreduce.application.classpath 和 yarn.application.classpath这两个参数，将其修改成绝对路径，这样就不会出现上述的错误。

（3）开始WordCount函数：

package wc;
import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.classification.InterfaceAudience.Public;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.record.compiler.JBoolean;

public class WCMapReduce 
	
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException
	
		Configuration conf=new Configuration();
		Job job=Job.getInstance(conf);
		job.setJobName("word count");
		job.setJarByClass(WCMapReduce.class);
		job.setJar("E:\\\\Ecplise\\\\WC.jar");
		//配置任务map和reduce类
		job.setMapperClass(WCMap.class);
		job.setReducerClass(WCReduce.class);
		//输出类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		//文件格式
		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		//设置输出输入路径
		FileInputFormat.addInputPath(job,new Path("hdfs://192.98.12.234:9000/Test/"));
		FileOutputFormat.setOutputPath(job, new Path("hdfs://192.98.12.234:9000/result"));
		//启动任务
		job.waitForCompletion(true);
	
	
	public static class WCMap extends Mapper<LongWritable, Text, Text, IntWritable>
	
		private static Text outKey=new Text();
		private static IntWritable outValue=new IntWritable(1);
		@Override
		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context)
				throws IOException, InterruptedException 
			// TODO Auto-generated method stub
			String words=value.toString();
			StringTokenizer tokenizer=new StringTokenizer(words,"\\\\s");
			while(tokenizer.hasMoreTokens())
			
				String word=tokenizer.nextToken();
				outKey.set(word);
				context.write(outKey, outValue);
			
		
	
	
	public static class WCReduce extends Reducer<Text, IntWritable, Text, IntWritable>
	
		private static IntWritable outValue=new IntWritable(); 
		@Override
		protected void reduce(Text arg0, Iterable<IntWritable> arg1,
				Reducer<Text, IntWritable, Text, IntWritable>.Context arg2) throws IOException, InterruptedException 
			// TODO Auto-generated method stub
			int sum=0;
			for(IntWritable i:arg1)
			
				sum+=i.get();
			
			outValue.set(sum);
			arg2.write(arg0,outValue);
		
	

需要注意的是，因为这里实现的是远程提交方法，所以在远程提交时需要将任务的jar包发送到集群中，但是ecplise中并没有自带这种框架，因此需要先将jar打好在相应的文件中，然后在程序中，通过下行代码指定jar的位置。

job.setJar("E:\\\\Ecplise\\\\WC.jar");

（4）配置提交任务的用户环境变量：

如果windows上的用户名称和linux上启动集群的用户名称不相同时，则需要添加一个环境变量来实现任务的提交：

（5）运行结果

16/03/30 21:09:14 INFO client.RMProxy: Connecting to ResourceManager at hadoop1/192.98.12.234:8032
16/03/30 21:09:14 WARN mapreduce.JobResourceUploader: Hadoop command-line option parsing not performed. Implement the Tool interface and execute your application with ToolRunner to remedy this.
16/03/30 21:09:14 INFO input.FileInputFormat: Total input paths to process : 1
16/03/30 21:09:14 INFO mapreduce.JobSubmitter: number of splits:1
16/03/30 21:09:15 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1459331173846_0031
16/03/30 21:09:15 INFO impl.YarnClientImpl: Submitted application application_1459331173846_0031
16/03/30 21:09:15 INFO mapreduce.Job: The url to track the job: http://hadoop1:8088/proxy/application_1459331173846_0031/
16/03/30 21:09:15 INFO mapreduce.Job: Running job: job_1459331173846_0031
16/03/30 21:09:19 INFO mapreduce.Job: Job job_1459331173846_0031 running in uber mode : false
16/03/30 21:09:19 INFO mapreduce.Job:  map 0% reduce 0%
16/03/30 21:09:24 INFO mapreduce.Job:  map 100% reduce 0%
16/03/30 21:09:28 INFO mapreduce.Job:  map 100% reduce 100%
16/03/30 21:09:29 INFO mapreduce.Job: Job job_1459331173846_0031 completed successfully
16/03/30 21:09:29 INFO mapreduce.Job: Counters: 49
	File System Counters
		FILE: Number of bytes read=19942
		FILE: Number of bytes written=274843
		FILE: Number of read operations=0
		FILE: Number of large read operations=0
		FILE: Number of write operations=0
		HDFS: Number of bytes read=15533
		HDFS: Number of bytes written=15671
		HDFS: Number of read operations=6
		HDFS: Number of large read operations=0
		HDFS: Number of write operations=2
	Job Counters 
		Launched map tasks=1
		Launched reduce tasks=1
		Data-local map tasks=1
		Total time spent by all maps in occupied slots (ms)=9860
		Total time spent by all reduces in occupied slots (ms)=2053
		Total time spent by all map tasks (ms)=2465
		Total time spent by all reduce tasks (ms)=2053
		Total vcore-seconds taken by all map tasks=2465
		Total vcore-seconds taken by all reduce tasks=2053
		Total megabyte-seconds taken by all map tasks=10096640
		Total megabyte-seconds taken by all reduce tasks=2102272
	Map-Reduce Framework
		Map input records=289
		Map output records=766
		Map output bytes=18404
		Map output materialized bytes=19942
		Input split bytes=104
		Combine input records=0
		Combine output records=0
		Reduce input groups=645
		Reduce shuffle bytes=19942
		Reduce input records=766
		Reduce output records=645
		Spilled Records=1532
		Shuffled Maps =1
		Failed Shuffles=0
		Merged Map outputs=1
		GC time elapsed (ms)=33
		CPU time spent (ms)=1070
		Physical memory (bytes) snapshot=457682944
		Virtual memory (bytes) snapshot=8013651968
		Total committed heap usage (bytes)=368050176
	Shuffle Errors
		BAD_ID=0
		CONNECTION=0
		IO_ERROR=0
		WRONG_LENGTH=0
		WRONG_MAP=0
		WRONG_REDUCE=0
	File Input Format Counters 
		Bytes Read=15429
	File Output Format Counters 
		Bytes Written=15671

因为MapReduce任务在src文件下配置那5个文件时，会在本地种启动任务。当任务在本地执行的，任务的名称中就会出现local，而上述的任务名称中并没有出现local，因此成功将任务提交到了Linux 集群中