Hadoop Distributed cache throws FileNotFound error

Question

Я пытаюсь использовать файл listOfWords для подсчета только тех слов из любого входного файла. Получение ошибки как FileNotFound, хотя я подтвердил, что файл находится в правильном месте в HDFS.

Внутри Driver:

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

Внутри Mapper: сообщение

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

вы можете попробовать это, чтобы получить файлы.

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

Вы можете перебирать файлы.

Answer 2

Попробуйте как 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); 
} 

В настройках Mapper()

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 3

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); 
    } 

Answer 4

В основной метод, я использую это.

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> 

Trick часть для меня использовать path.getName в FileReader, если не я FileNotFoundException