mapreduce卡住
- 作者: 爱上猫的诱惑
- 来源: 51数据库
- 2020-09-22
一、首先要知道此前提转载 若在windows的Eclipse工程中直接启动mapreduc程序，需要先把hadoop集群的配置目录下的xml都拷贝到src目录下，让程序自动读取集群的地址后去进行分布式运行(您也可以自己写java代码去设置job的configuration属性)。 若不拷贝，工程中bin目录没有完整的xml配置文件，则windows执行的mapreduce程序全部通过本机的jvm执行，作业名也是带有“local"字眼的作业，如job_local2062122004_0001。这不是真正的分布式运行mapreduce程序。 估计得研究org.apache.hadoop.conf.Configuration的源码，反正xml配置文件会影响执行mapreduce使用的文件系统是本机的windows文件系统还是远程的hdfs系统;还有影响执行mapreduce的mapper和reducer的是本机的jvm还是集群里面机器的jvm 二、本文的结论 第一点就是：windows上执行mapreduce，必须打jar包到所有slave节点才能正确分布式运行mapreduce程序。（我有个需求是要windows上触发一个mapreduce分布式运行） 第二点就是：Linux上，只需拷贝jar文件到集群master上,执行命令hadoopjarPackage.jarMainClassName即可分布式运行mapreduce程序。 第三点就是：推荐使用附一，实现了自动打jar包并上传，分布式执行的mapreduce程序。 附一、推荐使用此方法：实现了自动打jar包并上传，分布式执行的mapreduce程序： 请先参考博文五篇： Hadoop作业提交分析（一）~~（五） 引用博文的附件中EJob.java到你的工程中，然后main中添加如下方法和代码。 publicstaticFilecreatePack()throwsIOException{ FilejarFile=EJob.createTempJar("bin"); ClassLoaderclassLoader=EJob.getClassLoader(); Thread.currentThread().setContextClassLoader(classLoader); returnjarFile; } 在作业启动代码中使用打包： Jobjob=Job.getInstance(conf,"testAnaAction"); 添加： StringjarPath=createPack().getPath(); job.setJar(jarPath); 即可实现直接runasjavaapplication在windows跑分布式的mapreduce程序，不用手工上传jar文件。 附二、得出结论的测试过程 （未有空看书，只能通过愚笨的测试方法得出结论了） 一.直接通过windows上Eclipse右击main程序的java文件，然后"runasapplication"或选择hadoop插件"runonhadoop"来触发执行MapReduce程序的测试。 1，如果不打jar包到进集群任意linux机器上，它报错如下： [work]2012-06-2515:42:47,360-org.apache.hadoop.mapreduce.Job-10244[main]INFOorg.apache.hadoop.mapreduce.Job-map0%reduce0% [work]2012-06-2515:42:52,223-org.apache.hadoop.mapreduce.Job-15107[main]INFOorg.apache.hadoop.mapreduce.Job-TaskId:attempt_1403517983686_0056_m_000000_0,Status:FAILED Error:java.lang.RuntimeException:java.lang.ClassNotFoundException:ClassbookCount.BookCount$BookCountMappernotfound atorg.apache.hadoop.conf.Configuration.getClass(Configuration.java:1720) atorg.apache.hadoop.mapreduce.task.JobContextImpl.getMapperClass(JobContextImpl.java:186) atorg.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:721) atorg.apache.hadoop.mapred.MapTask.run(MapTask.java:339) atorg.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:162) atjava.security.AccessController.doPrivileged(NativeMethod) atjavax.security.auth.Subject.doAs(Subject.java:415) atorg.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1491) atorg.apache.hadoop.mapred.YarnChild.main(YarnChild.java:157) Causedby:java.lang.ClassNotFoundException:ClassbookCount.BookCount$BookCountMappernotfound atorg.apache.hadoop.conf.Configuration.getClassByName(Configuration.java:1626) atorg.apache.hadoop.conf.Configuration.getClass(Configuration.java:1718) 8more #Error:后重复三次 2012-06-2515:44:53,234-org.apache.hadoop.mapreduce.Job-37813[main]INFOorg.apache.hadoop.mapreduce.Job-map100%reduce100% 现象就是：报错，无进度，无运行结果。 2，拷贝jar包到“只是”集群master的$HADOOP_HOME/share/hadoop/mapreduce/目录上，直接通过windows的eclipse"runasapplication"和通过hadoop插件"runonhadoop"来触发执行，它报错同上。 现象就是：报错，无进度，无运行结果。 3，拷贝jar包到集群某些slave的$HADOOP_HOME/share/hadoop/mapreduce/目录上，直接通过windows的eclipse"runasapplication"和通过hadoop插件"runonhadoop"来触发执行 和报错： Error:java.lang.RuntimeException:java.lang.ClassNotFoundException:ClassbookCount.BookCount$BookCountMappernotfound atorg.apache.hadoop.conf.Configuration.getClass(Configuration.java:1720) atorg.apache.hadoop.mapreduce.task.JobContextImpl.getMapperClass(JobContextImpl.java:186) 和报错： Error:java.lang.RuntimeException:java.lang.ClassNotFoundException:ClassbookCount.BookCount$BookCountReducernotfound 现象就是：有报错，但仍然有进度，有运行结果。 4，拷贝jar包到集群所有slave的$HADOOP_HOME/share/hadoop/mapreduce/目录上，直接通过windows的eclipse"runasapplication"和通过hadoop插件"runonhadoop"来触发执行： 现象就是：无报错，有进度，有运行结果。 第一点结论就是：windows上执行mapreduce，必须打jar包到所有slave节点才能正确分布式运行mapreduce程序。 二在Linux上的通过以下命令触发MapReduce程序的测试。 hadoopjar$HADOOP_HOME/share/hadoop/mapreduce/bookCount.jarbookCount.BookCount 1，只拷贝到master，在master上执行。 现象就是：无报错，有进度，有运行结果。 2，拷贝随便一个slave节点,在slave上执行。 现象就是：无报错，有进度，有运行结果。 但某些节点上运行会报错如下，且运行结果。： 14/06/2516:44:02INFOmapreduce.JobSubmitter:Cleaningupthestagingarea/tmp/hadoop-yarn/staging/hduser/.staging/job_1403517983686_0071 Exceptioninthread"main"java.lang.NoSuchFieldError:DEFAULT_MAPREDUCE_APPLICATION_CLASSPATH atorg.apache.hadoop.mapreduce.v2.util.MRApps.setMRFrameworkClasspath(MRApps.java:157) atorg.apache.hadoop.mapreduce.v2.util.MRApps.setClasspath(MRApps.java:198) atorg.apache.hadoop.mapred.YARNRunner.createApplicationSubmissionContext(YARNRunner.java:443) atorg.apache.hadoop.mapred.YARNRunner.submitJob(YARNRunner.java:283) atorg.apache.hadoop.mapreduce.JobSubmitter.submitJobInternal(JobSubmitter.java:415) atorg.apache.hadoop.mapreduce.Job$10.run(Job.java:1268) atorg.apache.hadoop.mapreduce.Job$10.run(Job.java:1265) atjava.security.AccessController.doPrivileged(NativeMethod) atjavax.security.auth.Subject.doAs(Subject.java:415) atorg.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1491) atorg.apache.hadoop.mapreduce.Job.submit(Job.java:1265) atorg.apache.hadoop.mapreduce.Job.waitForCompletion(Job.java:1286) atcom.etrans.anaSpeed.AnaActionMr.run(AnaActionMr.java:207) atorg.apache.hadoop.util.ToolRunner.run(ToolRunner.java:70) atcom.etrans.anaSpeed.AnaActionMr.main(AnaActionMr.java:44) atsun.reflect.NativeMethodAccessorImpl.invoke0(NativeMethod) atsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) atsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) atjava.lang.reflect.Method.invoke(Method.java:606) atorg.apache.hadoop.util.RunJar.main(RunJar.java:212) 第二点结论就是：Linux上，只需拷贝jar文件到集群master上,执行命令hadoopjarPackage.jarMainClassName即可分布式运行mapreduce程序。
其实，将map处理的结果，传输到reduce上的过程，在mapreduce中，可以看做shuffle的过程。
在map端，每个map task都有一个内存缓冲区，存储着map的输出结果，当缓冲区快满的时候需要将缓冲区的数据以一个临时文件的方式存放到磁盘，当整个map task结束后再对磁盘中这个map task产生的所有临时文件做合并，生成最终的正式输出文件，然后等待reduce task来拉数据。
a. 在map task执行时，它的输入数据来源于hdfs的block，当然在mapreduce概念中，map task只读取split。split与block的对应关系可能是多对一，默认是一对一。
b. 在经过mapper的运行后，我们得知mapper的输出是这样一个key/value对。到底当前的key应该交由哪个reduce去做呢，是需要现在决定的。 mapreduce提供partitioner接口，它的作用就是根据key或value及reduce的数量来决定当前的这对输出数据最终应该交由哪个reduce task处理。默认对key hash后再以reduce task数量取模。默认的取模方式只是为了平均reduce的处理能力，如果用户自己对partitioner有需求，可以订制并设置到job上。接下来，需要将数据写入内存缓冲区中，缓冲区的作用是批量收集map结果，减少磁盘io的影响。我们的key/value对以及partition的结果都会被写入缓冲区。当然写入之前，key与value值都会被序列化成字节数组。
c. 这个内存缓冲区是有大小限制的，默认是100mb。当map task的输出结果很多时，就可能会撑爆内存，所以需要在一定条件下将缓冲区中的数据临时写入磁盘，然后重新利用这块缓冲区。这个从内存往磁盘写数据的过程被称为spill，中文可译为溢写，字面意思很直观。这个溢写是由单独线程来完成，不影响往缓冲区写map结果的线程。溢写线程启动时不应该阻止map的结果输出，所以整个缓冲区有个溢写的比例spill.percent。这个比例默认是0.8，也就是当缓冲区的数据已经达到阈值（buffer size * spill percent = 100mb * 0.8 = 80mb），溢写线程启动，锁定这80mb的内存，执行溢写过程。map task的输出结果还可以往剩下的20mb内存中写，互不影响。
当溢写线程启动后，需要对这80mb空间内的key做排序(sort)。排序是mapreduce模型默认的行为，这里的排序也是对序列化的字节做的排序。 存缓冲区没有对将发送到相同reduce端的数据做合并，那么这种合并应该是体现是磁盘文件中的。从官方图上也可以看到写到磁盘中的溢写文件是对不同的reduce端的数值做过合并。所以溢写过程一个很重要的细节在于，如果有很多个key/value对需要发送到某个reduce端去，那么需要将这些key/value值拼接到一块，减少与partition相关的索引记录。
其实，将map处理的结果，传输到reduce上的过程，在mapreduce中，可以看做shuffle的过程。
在map端，每个map task都有一个内存缓冲区，存储着map的输出结果，当缓冲区快满的时候需要将缓冲区的数据以一个临时文件的方式存放到磁盘，当整个map task结束后再对磁盘中这个map task产生的所有临时文件做合并，生成最终的正式输出文件，然后等待reduce task来拉数据。
a. 在map task执行时，它的输入数据来源于hdfs的block，当然在mapreduce概念中，map task只读取split。split与block的对应关系可能是多对一，默认是一对一。
b. 在经过mapper的运行后，我们得知mapper的输出是这样一个key/value对。到底当前的key应该交由哪个reduce去做呢，是需要现在决定的。 mapreduce提供partitioner接口，它的作用就是根据key或value及reduce的数量来决定当前的这对输出数据最终应该交由哪个reduce task处理。默认对key hash后再以reduce task数量取模。默认的取模方式只是为了平均reduce的处理能力，如果用户自己对partitioner有需求，可以订制并设置到job上。接下来，需要将数据写入内存缓冲区中，缓冲区的作用是批量收集map结果，减少磁盘io的影响。我们的key/value对以及partition的结果都会被写入缓冲区。当然写入之前，key与value值都会被序列化成字节数组。
c. 这个内存缓冲区是有大小限制的，默认是100mb。当map task的输出结果很多时，就可能会撑爆内存，所以需要在一定条件下将缓冲区中的数据临时写入磁盘，然后重新利用这块缓冲区。这个从内存往磁盘写数据的过程被称为spill，中文可译为溢写，字面意思很直观。这个溢写是由单独线程来完成，不影响往缓冲区写map结果的线程。溢写线程启动时不应该阻止map的结果输出，所以整个缓冲区有个溢写的比例spill.percent。这个比例默认是0.8，也就是当缓冲区的数据已经达到阈值（buffer size * spill percent = 100mb * 0.8 = 80mb），溢写线程启动，锁定这80mb的内存，执行溢写过程。map task的输出结果还可以往剩下的20mb内存中写，互不影响。
当溢写线程启动后，需要对这80mb空间内的key做排序(sort)。排序是mapreduce模型默认的行为，这里的排序也是对序列化的字节做的排序。 存缓冲区没有对将发送到相同reduce端的数据做合并，那么这种合并应该是体现是磁盘文件中的。从官方图上也可以看到写到磁盘中的溢写文件是对不同的reduce端的数值做过合并。所以溢写过程一个很重要的细节在于，如果有很多个key/value对需要发送到某个reduce端去，那么需要将这些key/value值拼接到一块，减少与partition相关的索引记录。
推荐阅读
