如何分布式运行mapreduce程序

一、 首先要知道此前提 转载

若在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上,执行命令hadoop jarPackage.jar MainClassName即可分布式运行mapreduce程序。

第三点就是： 推荐使用附一，实现了自动打jar包并上传，分布式执行的mapreduce程序。

附一、 推荐使用此方法：实现了自动打jar包并上传，分布式执行的mapreduce程序：

请先参考博文五篇：

Hadoop作业提交分析（一）~~（五）

引用博文的附件中EJob.java到你的工程中，然后main中添加如下方法和代码。

public static File createPack() throws IOException {

File jarFile = EJob.createTempJar("bin")

ClassLoader classLoader = EJob.getClassLoader()

Thread.currentThread().setContextClassLoader(classLoader)

return jarFile

}

在作业启动代码中使用打包：

Job job = Job.getInstance(conf, "testAnaAction")

添加：

String jarPath = createPack().getPath()

job.setJar(jarPath)

即可实现直接run as java application 在windows跑分布式的mapreduce程序，不用手工上传jar文件。

附二、得出结论的测试过程

（未有空看书，只能通过愚笨的测试方法得出结论了）

一. 直接通过windows上Eclipse右击main程序的java文件，然后"run as application"或选择hadoop插件"run on hadoop"来触发执行MapReduce程序的测试。

1，如果不打jar包到进集群任意linux机器上，它报错如下：

[work] 2012-06-25 15:42:47,360 - org.apache.hadoop.mapreduce.Job -10244 [main] INFO org.apache.hadoop.mapreduce.Job - map 0% reduce 0%

[work] 2012-06-25 15:42:52,223 - org.apache.hadoop.mapreduce.Job -15107 [main] INFO org.apache.hadoop.mapreduce.Job - Task Id : attempt_1403517983686_0056_m_000000_0, Status : FAILED

Error: java.lang.RuntimeException: java.lang.ClassNotFoundException: Class bookCount.BookCount$BookCountMapper not found

at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:1720)

at org.apache.hadoop.mapreduce.task.JobContextImpl.getMapperClass(JobContextImpl.java:186)

at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:721)

at org.apache.hadoop.mapred.MapTask.run(MapTask.java:339)

at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:162)

at java.security.AccessController.doPrivileged(Native Method)

at javax.security.auth.Subject.doAs(Subject.java:415)

at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1491)

at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:157)

Caused by: java.lang.ClassNotFoundException: Class bookCount.BookCount$BookCountMapper not found

at org.apache.hadoop.conf.Configuration.getClassByName(Configuration.java:1626)

at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:1718)

... 8 more

# Error:后重复三次

2012-06-25 15:44:53,234 - org.apache.hadoop.mapreduce.Job -37813 [main] INFO org.apache.hadoop.mapreduce.Job - map 100% reduce 100%

现象就是：报错，无进度，无运行结果。

2，拷贝jar包到“只是”集群master的$HADOOP_HOME/share/hadoop/mapreduce/目录上，直接通过windows的eclipse "run as application"和通过hadoop插件"run on hadoop"来触发执行，它报错同上。

现象就是：报错，无进度，无运行结果。

3，拷贝jar包到集群某些slave的$HADOOP_HOME/share/hadoop/mapreduce/目录上，直接通过windows的eclipse "run as application"和通过hadoop插件"run on hadoop"来触发执行

和报错：

Error: java.lang.RuntimeException: java.lang.ClassNotFoundException: Class bookCount.BookCount$BookCountMapper not found

at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:1720)

at org.apache.hadoop.mapreduce.task.JobContextImpl.getMapperClass(JobContextImpl.java:186)

和报错：

Error: java.lang.RuntimeException: java.lang.ClassNotFoundException: Class bookCount.BookCount$BookCountReducer not found

现象就是：有报错，但仍然有进度，有运行结果。

4，拷贝jar包到集群所有slave的$HADOOP_HOME/share/hadoop/mapreduce/目录上，直接通过windows的eclipse "run as application"和通过hadoop插件"run on hadoop"来触发执行：

现象就是：无报错，有进度，有运行结果。

第一点结论就是： windows上执行mapreduce，必须打jar包到所有slave节点才能正确分布式运行mapreduce程序。

二 在Linux上的通过以下命令触发MapReduce程序的测试。

hadoop jar $HADOOP_HOME/share/hadoop/mapreduce/bookCount.jar bookCount.BookCount

1，只拷贝到master，在master上执行。

现象就是：无报错，有进度，有运行结果。

2，拷贝随便一个slave节点,在slave上执行。

现象就是：无报错，有进度，有运行结果。

但某些节点上运行会报错如下，且运行结果。：

14/06/25 16:44:02 INFO mapreduce.JobSubmitter: Cleaning up the staging area /tmp/hadoop-yarn/staging/hduser/.staging/job_1403517983686_0071

Exception in thread "main" java.lang.NoSuchFieldError: DEFAULT_MAPREDUCE_APPLICATION_CLASSPATH

at org.apache.hadoop.mapreduce.v2.util.MRApps.setMRFrameworkClasspath(MRApps.java:157)

at org.apache.hadoop.mapreduce.v2.util.MRApps.setClasspath(MRApps.java:198)

at org.apache.hadoop.mapred.YARNRunner.createApplicationSubmissionContext(YARNRunner.java:443)

at org.apache.hadoop.mapred.YARNRunner.submitJob(YARNRunner.java:283)

at org.apache.hadoop.mapreduce.JobSubmitter.submitJobInternal(JobSubmitter.java:415)

at org.apache.hadoop.mapreduce.Job$10.run(Job.java:1268)

at org.apache.hadoop.mapreduce.Job$10.run(Job.java:1265)

at java.security.AccessController.doPrivileged(Native Method)

at javax.security.auth.Subject.doAs(Subject.java:415)

at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1491)

at org.apache.hadoop.mapreduce.Job.submit(Job.java:1265)

at org.apache.hadoop.mapreduce.Job.waitForCompletion(Job.java:1286)

at com.etrans.anaSpeed.AnaActionMr.run(AnaActionMr.java:207)

at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:70)

at com.etrans.anaSpeed.AnaActionMr.main(AnaActionMr.java:44)

at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)

at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)

at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)

at java.lang.reflect.Method.invoke(Method.java:606)

at org.apache.hadoop.util.RunJar.main(RunJar.java:212)

第二点结论就是： Linux上，只需拷贝jar文件到集群master上,执行命令hadoop jarPackage.jar MainClassName即可分布式运行mapreduce程序。

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

---