Hadoop Job的提交,到底提交了什么?

##一个最基本的Hadoo任务 WordCountMapper:

<pre class="prettyprint lang-java"> private final static IntWritable one = new IntWritable(1); @Override public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { StringTokenizer tokenizer = new StringTokenizer(value.toString()); while (tokenizer.hasMoreTokens()) { String str = tokenizer.nextToken(); context.write(new Text(StringUtils.trim(str.replaceAll("\\W", ""))), one); } } </pre>

WordCountReduce:

<pre class="prettyprint lang-java"> @Override public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } context.write(key, new IntWritable(sum)); } </pre>

WordCount:

<pre class="prettyprint lang-java"> final Configuration conf = new Configuration(); Job job = new Job(conf); job.setJobName("wordcount"); job.setJarByClass(WordCount.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); job.setMapperClass(WordCountMapper.class); job.setReducerClass(WordCountReduce.class); job.setInputFormatClass(TextInputFormat.class); job.setOutputFormatClass(TextOutputFormat.class); FileInputFormat.setInputPaths(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); boolean success = job.waitForCompletion(true); </pre>

如你所见, 这是个Hadoop基础的入门例子, 若是你了解Hadoop, 你已经对这些代码熟记于心了.这篇文章我想说明Hadoop提交Job到底提交了那些东西,提交了哪些类,不须要提交那些东西.

##Hadoop任务提交

###传统的Hadoop任务提交

把上面的代码<code> WordCountMapper, WordCountReduce, WordCount</code>打包成jar,放到hadoop目录下, 使用<code>hadoop jar wordcount.jar WordCount</code>运行任务. 这样方式我称为传统的方式,也是《hadoop权威指南》上一向的方法.

###Eclipse的hadoop插件的Hadoop任务提交

若是你开发过Hadoop的Job, 那么对这个应该很熟悉.大多数开发测试都是用这个提交任务的,若是每次都是打包成jar, 再用hadoop jar ** 这还不把人搞疯.

若是你还细心,你会发现,你选好Hadoop的jobtracker,提交任务的前一刻,Eclipse会弹出一个浮动窗口,上面跳动着显示不少jar名.为何会这样?它作了什么?

###在Eclipse中当作Java Application运行为何不能够?

Hadoop的job项目都有main方法,这个是符合JavaApplication运行条件的,那么咱们是否是可使用Eclipse中直接运行呢?当咱们尝试运行的时候,程序是能够运行的,但总当运行一下子(几秒钟)后抛出WordCountMapper <code>ClassNotFount</code>的错误.

那么为何程序不是直接抛出错误而是过了一下子才抛出?为何用Eclipse Hadoop插件运行不会发生这个错误.

##背景

写这篇文章前,咱们已经正在开发一个Hadoop的任务调度的系统. 也就是一个项目中提早写好不少个Hadoop Job绑定到里边,若是想要运行哪一个Job,咱们就从前台配置好参数,并把这个Job提交到Hadoop集群. 并从前台不断的获得Job的运行任务信息,取得Job的执行进度. 不关心进度的话直接去喝茶就能够,完了来看结果.

若是每次都是<code>hadoop jar wordcount.jar WordCount</code>未免太弱智. 若是每次从Eclispe中运行,那专业性太强,也不可取.是否是有更好的方法提交任务?

答案确定是有的.

##用JVisualVM监视Eclipse hadoop插件的Hadoop任务提交

打开JVisualVM准备着,运行一个Job. 运行后当即就能够看到一个Java进程. 用JVisualVM打开这个进程查看,如图:

我打开我电脑上的目录<code>F:\Eclipse\workspace.metadata.plugins\org.apache.hadoop.eclipse\hadoop-conf-1007657720166395816</code>而且查看了它的上级目录,顿时一些皆明朗了.Eclipse Hadoop插件居然把个人项目下全部的类,资源文件打包成jar而后运行的.

##使用Hadoop Api提交Job,完美解决方案

其实我也是从这篇文章(http://luliangy.iteye.com/blog/1401453)中找到灵感的.为何把项目打包,以Java Application的方式就正常运行了.

<pre class="prettyprint lang-java"> ((JobConf) job.getConfiguration()).setJar("wordcount.jar"); job.setJarByClass(WordCount.class); </pre>

冲着这股劲我看了不少Hadoop API,终于找到为何.

经过<code>job.setJarByClass(WordCount.class); </code>这条路查看源码, 你会找到以下的两个方法.

<pre class="prettyprint lang-java"> private static String findContainingJar(Class my_class) { ClassLoader loader = my_class.getClassLoader(); String class_file = my_class.getName().replaceAll("\\.", "/") + ".class"; try { for(Enumeration itr = loader.getResources(class_file); itr.hasMoreElements();) { URL url = (URL) itr.nextElement(); if ("jar".equals(url.getProtocol())) { String toReturn = url.getPath(); if (toReturn.startsWith("file:")) { toReturn = toReturn.substring("file:".length()); } toReturn = URLDecoder.decode(toReturn, "UTF-8"); return toReturn.replaceAll("!.*$", ""); } } } catch (IOException e) { throw new RuntimeException(e); } return null; } public void setJarByClass(Class cls) { String jar = findContainingJar(cls); if (jar != null) { setJar(jar); } } </pre>

我说说<code>findContainingJar</code>有什么做用? 当使用<code>job.setJarByClass(WordCount.class);</code>设置类的时候, Hadoop Client能从你的classpath中取得<code>WordCount.class</code>所在的jar包的jar File绝对路径.若是找不到jar, <code>setJar(jar);</code>方法没有执行,jar确定是个空值.

咱们在Eclipse中直接以Java Application运行的时候,classpath是一个本地文件夹, findContainingJar确定找不到项目的jar.也就是Mapper和Reduce所在的jar. 这样在提交任务时候Configuration中<code>mapper.jar</code>属性是一个空值.这也就解释了为何在Eclipse中当作Java Application运行时老是过一段时间后才发生ClassNotFound的错误缘由.

其实到这里Hadoop提交了什么也好解释了.

Hadoop向集群中提交了一个xml和一个携带Mapper/Reduce的jar. xml就是Configuration对象序列化的结果.

说到这里也许你已经发现,这是一个开发上的架构问题.既然Hadoop Job须要Map/Reduce的jar.咱们应该把全部的Map/Reduce单独在一个项目中开发.而后打包放入调度系统项目的ClassPath就行了.而后在调度系统中构造Job,并把<code>job.setJarByClass(class);</code>中的class设置为该Job的map clas或者reduce class就好了.

##哪些是在Client执行的?哪些是在Hadoop集群中执行?

一个Hadoop 任务通常都有3个类(Map/Reduce/Job).<code>WordCountMapper, WordCountReduce, WordCount</code>你认为这三个类都会提交到集群中执行吗?

不是! 只有Mapper和Reduce这2个类会提交到Hadoop集群, MapReduce执行也是这2个类. WordCount只是充当一个配置Job的客户端,而且提交任务,以后又定时轮询Job的运行状态输出简单的日志,直到任务完成,WordCount的这个进程会自动退出.

##Hadoop分布式缓存

讲到这里你也许能顺利的实现一个和我相同思路的系统了.

可是我仍是想说一个常见错误. 不是Mapper Class NotFound,而是Mapper中使用的Class NotFound. 而你又不想往hadoop集群中添加jar包,也不想重启Hadoop集群. 你可使用Hadoop提供的一个类:<code>DistributedCache</code>

1. DistributedCache.addArchiveToClassPath() 添加hdfs上的jar到MapReduce的classpath
2. DistributedCache.addCacheFile(new URI("/myapp/lookup.dat#lookup.dat"), job);
3. DistributedCache.addCacheArchive(new URI("/myapp/map.zip", job);
4. DistributedCache.addFileToClassPath(new Path("/myapp/mylib.jar"), job);
5. DistributedCache.addCacheArchive(new URI("/myapp/mytar.tar", job);
6. DistributedCache.addCacheArchive(new URI("/myapp/mytgz.tgz", job);
7. DistributedCache.addCacheArchive(new URI("/myapp/mytargz.tar.gz", job);