Hadoop分布式缓存抛出FileNotFound错误

Question

我正在尝试使用 listOfWords 文件来仅计算任何输入文件中的那些单词。即使我已经验证该文件在 HDFS 中的正确位置，也会出现 FileNotFound 错误。

内部驱动：

    Configuration conf = new Configuration();
    DistributedCache.addCacheFile(new URI("/user/training/listOfWords"), conf);
    Job job = new Job(conf,"CountEachWord Job");

内部映射器：

private Path[] ref_file;
ArrayList<String> globalList = new ArrayList<String>();

public void setup(Context context) throws IOException{

    this.ref_file = DistributedCache.getLocalCacheFiles(context.getConfiguration());

    FileSystem fs = FileSystem.get(context.getConfiguration());

    FSDataInputStream in_file = fs.open(ref_file[0]);
    System.out.println("File opened");

    BufferedReader br  = new BufferedReader(new InputStreamReader(in_file));//each line of reference file
    System.out.println("BufferReader invoked");

    String eachLine = null;
    while((eachLine = br.readLine()) != null)
    {
        System.out.println("eachLine is: "+ eachLine);
        globalList.add(eachLine);

    }

}

错误信息：

 hadoop jar CountOnlyMatchWords.jar CountEachWordDriver Rhymes CountMatchWordsOut1
 Warning: $HADOOP_HOME is deprecated.

14/10/07 22:28:59 WARN mapred.JobClient: Use GenericOptionsParser for parsing the     arguments.      Applications should implement Tool for the same.
14/10/07 22:28:59 INFO input.FileInputFormat: Total input paths to process : 1
14/10/07 22:28:59 INFO util.NativeCodeLoader: Loaded the native-hadoop library
14/10/07 22:28:59 WARN snappy.LoadSnappy: Snappy native library not loaded
14/10/07 22:29:00 INFO mapred.JobClient: Running job: job_201409300531_0041
14/10/07 22:29:01 INFO mapred.JobClient:  map 0% reduce 0%
14/10/07 22:29:14 INFO mapred.JobClient: Task Id : attempt_201409300531_0041_m_000000_0, Status : FAILED
 java.io.FileNotFoundException: File does not exist: /home/training/hadoop-temp/mapred/local /taskTracker/distcache/5910352135771601888_2043607380_1633197895/localhost/user/training/listOfWords

我已验证上述文件存在于 HDFS 中。我也尝试使用 localRunner。仍然没有工作。

Answer 1

在main方法中，我使用了这个。

  Job job = Job.getInstance();
  job.setJarByClass(DistributedCacheExample.class);
  job.setJobName("Distributed cache example");
  job.addCacheFile(new Path("/user/cloudera/datasets/abc.dat").toUri());

然后在 Mapper 中我使用了这个样板。

  protected void setup(Context context) throws IOException, InterruptedException {
     URI[] files = context.getCacheFiles();
     for(URI file : files){
     if(file.getPath().contains("abc.dat")){
       Path path = new Path(file);
       BufferedReader reader = new BufferedReader(new FileReader(path.getName()));
       String line = reader.readLine();
       while(line != null){
         ......
       }
     }
  }

我正在处理这些依赖项

  <dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>hadoop-common</artifactId>
  <version>2.7.3</version>
  </dependency>

  <dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>hadoop-mapreduce-client-core</artifactId>
  <version>2.7.3</version>
  </dependency>

对我来说，诀窍是在FileReader 中使用path.getName，否则我会得到FileNotFoundException

Answer 2

你可以试试这个来检索文件。

URI[] 文件 = DistributedCache.getCacheFiles(context.getConfiguration());

您可以遍历文件。

Answer 3

试试this

在驱动程序中

Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(conf);
Path cachefile = new Path("path/to/file");
FileStatus[] list = fs.globStatus(cachefile);
for (FileStatus status : list) {
 DistributedCache.addCacheFile(status.getPath().toUri(), conf);
}

在映射器设置（）中

public void setup(Context context) throws IOException{
 Configuration conf = context.getConfiguration();
 FileSystem fs = FileSystem.get(conf);
 URI[] cacheFiles = DistributedCache.getCacheFiles(conf);
 Path getPath = new Path(cacheFiles[0].getPath());  
 BufferedReader bf = new BufferedReader(new InputStreamReader(fs.open(getPath)));
 String setupData = null;
 while ((setupData = bf.readLine()) != null) {
   System.out.println("Setup Line in reducer "+setupData);
 }
}

Answer 4

   try {
        URI[] cacheFiles = DistributedCache.getCacheFiles(job); // Fetch the centroid file from distributed cache
        Path getPath = new Path(cacheFiles[0].getPath());  
        FileSystem fs = FileSystem.get(job);
        if (cacheFiles != null && cacheFiles.length > 0) {
            // Goes in if the file exist and is not empty
            String line; 
            centers.clear(); // clearing the centers array list each time
            BufferedReader cacheBufferReader = new BufferedReader(new InputStreamReader(fs.open(getPath)));
            try {
                while ((line = cacheBufferReader.readLine()) != null) {
                        centers.add(line);
                } 
            } catch (IOException e) {
                System.err.println("Exception: " + e);
            }
        }
    } catch (IOException e) {
        System.err.println("Exception: " + e);
    }