[转]MapReduce程式调用第三方包和本地库

MapReduce程式调用第三方包和本地库
-------------------------
问题：
在MP程式中如何在不同的TaskTracker节点上调用第三方jar包和读取一些只读的文件。


解决方法介绍：
我们知道，在Hadoop中有一个叫做DistributedCache的东东，它是用来分发应用特定的只读文件和一个jar包的，以供Map-Reduce框架在启动任务和运行的时候使用这些缓冲的文件或者是把第三方jar包添加到其classpath路径中去，要注意的是DistributedCache的使用是有一个前提的，就它会认为这些通过urls来表示的文件已经在hdfs文件系统里面，所以这里在使用的时候第一步就是要把这些文件上传到HDFS中。


然后Hadoop框架会把这些应用所需要的文件复制到每个准备启动的节点上去，它会把这些复制到mapred.temp.dir配置的目录中去，以供相应的Task节点使用。


这里要注意的DistriubtedCache分发的文件分成公有与私有文件，公有文件可以给HDFS中的所有用户使用，而私有文件只能被特定的用户所使用，用户可以配置上传文件的访问权限来达到这种效果。


DistributeCache的使用一般分成三步：
1. 配置应用程序的cache,把需要使用的文件上传到DFS中去
   
[html] view plaincopy
$ bin/hadoop fs -copyFromLocal lookup.dat /myapp/lookup.dat   
$ bin/hadoop fs -copyFromLocal map.zip /myapp/map.zip   
$ bin/hadoop fs -copyFromLocal mylib.jar /myapp/mylib.jar 
$ bin/hadoop fs -copyFromLocal mytar.tar /myapp/mytar.tar 
$ bin/hadoop fs -copyFromLocal mytgz.tgz /myapp/mytgz.tgz 
$ bin/hadoop fs -copyFromLocal mytargz.tar.gz /myapp/mytargz.tar.gz 


2. 配置JobConf

[html] view plaincopy
JobConf job = new JobConf(); 
DistributedCache.addCacheFile(new URI("/myapp/lookup.dat#lookup.dat"),job); // 这里的lookup.dat加了一个符号连接 
DistributedCache.addCacheArchive(new URI("/myapp/map.zip", job); 
DistributedCache.addFileToClassPath(new Path("/myapp/mylib.jar"), job); // 这里是把相应的jar包加到Task的启动路径上去 
DistributedCache.addCacheArchive(new URI("/myapp/mytar.tar", job); 
DistributedCache.addCacheArchive(new URI("/myapp/mytgz.tgz", job); 
DistributedCache.addCacheArchive(new URI("/myapp/mytargz.tar.gz", job); 




3. 在Mapper或者Reducer任务中使用这些文件
   
[html] view plaincopy
public static class MapClass extends MapReduceBase   
    implements Mapper<K, V, K, V> { 
     
      private Path[] localArchives; 
      private Path[] localFiles; 
       
      public void configure(JobConf job) { 
        // Get the cached archives/files 
        localArchives = DistributedCache.getLocalCacheArchives(job);  // 得到本地打包的文件，一般是数据文件，如字典文件 
        localFiles = DistributedCache.getLocalCacheFiles(job);        // 得到本地缓冲的文件，一般是配置文件等 
      } 
       
      public void map(K key, V value,  
                      OutputCollector<K, V> output, Reporter reporter)  
      throws IOException { 
        // Use data from the cached archives/files here 
        // ... 
        // ... 
        output.collect(k, v); 
      } 
    } 




使用新的MP接口要注意的地方：
1. 我们知道，新的MP接口使用了Job这个类来对MP任务进行配置，这里使用的时候要注意一点
   Configuration conf = new Configuration();
   // 对conf加入配置信息  - 正确方法
   Job job = new Job(conf,"word count");
   // 对conf加入配置信息 - 这是有问题的，这些配置不会生效，因为这里生成Job的时候它会对conf进行复制，这个看一下Job的源代码就知道。
   // 这里可以用job.getConfiguration()来得到其内部的conf对象，这样就不会有问题。


2. 如果你在启动MP任务之前调用了第三方jar包的类，那这就会有问题，会在启动任务的时候找不到这个类。这个问题我还没有找到好的解决办法，一个办法就是把这些类想办法移到MP任务中，如果有朋友知道更加好的办法，请告诉我一下，多谢了。我感觉Nutch中也会有同样的问题，什么时候研究一下Nutch的代码，说不定会有很多关于Hadoop方面的收获。


参考：
1. http://hadoop.apache.org/common/docs/current/api/org/apache/hadoop/filecache/DistributedCache.html
2. http://hadoop.apache.org/common/docs/current/mapred_tutorial.html#IsolationRunner