Hadoop 中使用DistributedCache遇到的有关问题

Hadoop 中使用DistributedCache遇到的问题

?

自己在写MAR/REDUCE代码时，遇到了一个问题，一个大数据文件和一个小数据文件匹配计算，但是小数据文件太小，所以想采用HIVE的MAP JOIN的方式，把小数据文件放到直接大数据文件map的datanode的内存中，这样少了MR代码的1对N的数据文件关联。
实现这个的最佳方案就是利用distributed cache。HIVE的MAP JOIN也是利用这个技术。
首先简要介绍一下distributed cache是如何使用的，然后总结下自己在使用distributed cache遇到的问题，这些问题网上也有人遇到，但是没有给出明确的解释。希望能够帮助同样遇到此类问题的朋友。
distributed cache至少有如下的两类类应用：

1.MAP、REDUCE本身和之间共享的较大数据量的数据2.布置第三方JAR包，可以避免集群的删减导致部分依赖的机器的JAR包的丢失

distributed cache使用的流程总结如下：

1.在HDFS上准备好要共享的数据(text、archive、jar)2.在distributed cache中添加文件3.在mapper或者reducer类中获取数据4.在map或者reduce函数中使用数据

1.数据本来就在HDFS上，所以省去了流程中的第一步可以使用hadoop fs?-copyFromLocal把本地文件cp到HDFS上2.在distributed cache中添加文件

public static void disCache(String dimDir, JobConf conf) throws IOException {           FileSystem fs = FileSystem.get(URI.create(dimDir), conf);           FileStatus[] fileDir = fs.listStatus(new Path(dimDir));           for (FileStatus file : fileDir) {                DistributedCache.addCacheFile(URI.create(file.getPath().toString()), conf);           }}

因为我利用的数据是HIVE脚本生成的，所以无法指定具体的文件路径，采用这种方式把一张HIVE表的所有数据都加载到cache中。如果能直接明确知道文件名称就简单很多了，例如：

DistributedCache.addCacheFile(URI.create(“/mytestfile/file.txt”), conf);

3.在mapper或者reducer中获取数据

public void configure(JobConf job) {          try {               urlFile = DistributedCache.getLocalCacheFiles(job);               seoUrlFile = "file://" + urlFile[0].toString();               allUrlFile = "file://" + urlFile[1].toString();               getData(seoUrlFile, seoUrlDim);               getData(allUrlFile, allUrlDim);          } catch (IOException e) {                     // TODO Auto-generated catch block               e.printStackTrace();           }}

?? ? ?我在流程2中添加了两个文件，通过DistributedCache.getLocalCacheFiles(job)获取Path[]，针对不同文件调用getData，把文件数据放在不同的List中seoUrlDim，allUrlDim? ? ?如此一来，distributed cache的过程就结束了，接下来就在map()或者reduce()中使用这些数据就OK了。4.在map或者reduce函数中使用数据? ? ?一定要区分流程3和4，3是获取数据，4是使用数据。我就是在前期没弄明白这个的差异，导致内存溢出。
虽说好像挺简单的，但是在实现这个代码的过程中有如下几个问题困扰了我好久，网上也没找到很好的解决方案，后来在兄弟的帮助下搞定

1.FileNotFoundException

2.java.lang.OutOfMemoryError: GC overhead limit exceeded

1.FileNotFoundException? ? ?这个问题涉及到DistributedCache.getLocalCacheFiles(job) 这个函数，此函数返回的Path[]是执行map或者reduce的datanode的本地文件系统中路径，但是我在getData中利用的SequenceFile.Reader的默认filesystem是hdfs，这就导致获取数据时是从hdfs上找文件，但是这个文件是在本地文件系统中，所以就会报出这个错误FileNotFoundException。? ? ?例如，DistributedCache.getLocalCacheFiles(job)返回的PATH路径是：/home/dwapp/hugh.wangp/mytestfile/file.txt，在默认文件系统是hdfs时，获取数据时会读hdfs://hdpnn:9000/home/dwapp/hugh.wangp/mytestfile/file.txt，但是我们是在本地文件系统，为了避免数据获取函数选择"错误"的文件系统，我们在/home/dwapp/hugh.wangp/mytestfile/file.txt前加上"file://"，这样就会从本地文件系统中读取数据了。就像我例子中的seoUrlFile?=?"file://"?+?urlFile[0].toString();
2.java.lang.OutOfMemoryError: GC overhead limit exceeded? ? ?这个问题是我在没搞明白如何使用distributed cache时犯下的错误，也就是我没弄明白流程3和4的区别。我把流程3中说的获取数据的过程放在map函数中，而在map函数中其实是使用数据的过程。这个错误因为使每用一个map就要获取一下数据，也就是初始化一个list容器，使一个datanode上起N个map，就要获取N个list容器，内容溢出也就是自然而然的事情了。? ? ?我们一定要把获取数据的过程放在mapper或reducer类的configure()函数中，这样对应一个datanode就只有一份数据，N个map可以共享着一份数据。

内容单薄，望志同道合之士互相学习。 1 楼 neyshule 2012-06-10   java.io.FileNotFoundException: File /juker/cm does not exist.
你好，我也出现了相同的错误，请问怎么打印才能找出这个错呢？
2 楼 hugh.wangp 2012-06-10   可以直接在获取URL的前后System.out.println()打印些标签定位错误

查看更多 下一篇