如何在Hadoop Map/Reduce作业中访问DistributedCache？

Question

我想要一个小的文件传递给我使用的运行作业GenericOptionsParser的-files标志:

$ hadoop jar MyJob.jar -conf /path/to/cluster-conf.xml -files /path/to/local-file.csv data/input data/output

这应该将作业发送到我的集群并附加local-file.csv以在需要时可用于Mapper/Reducer.当我在伪分布式模式下运行它时,它工作得很好,但是当我在集群上启动作业时,似乎无法找到该文件.我正在使用mapper的setup方法读取文件,如下所示:

public static class TheMapper extends Mapper<LongWritable, Text, Text, Text> {

  @Override
  public void setup(Context context) throws IOException, InterruptedException {

    URI[] uriList = DistributedCache.getCacheFiles( context.getConfiguration() );
    CsvReader csv = new CsvReader(uriList[0].getPath());

    // work with csv file..
  }

  // ..
}

当作业运行时,我得到以下异常:

java.io.FileNotFoundException: File /hdfs/tmp/mapred/staging/hduser/.staging/job_201205112311_011/files/local-file.csv does not exist.
at com.csvreader.CsvReader.<init>(Unknown Source)
at com.csvreader.CsvReader.<init>(Unknown Source)
at com.csvreader.CsvReader.<init>(Unknown Source)
at MyJob$TheMapper.setup(MyJob.java:167)
at org.apache.hadoop.mapreduce.Mapper.run(Mapper.java:142)
...

知道我做错了什么吗？谢谢.

Answer 1

这是一个常见问题 - 该-files选项与DistributedCache不同.

使用时-files,GenericOptionsParser配置一个名为的作业属性tmpfiles,而DistributedCache使用一个名为的属性mapred.cache.files.

此外,分布式缓存期望文件已经在HDFS中并将它们复制到任务节点,其中-files在作业提交时将文件复制到HDFS,然后将它们复制到每个任务节点.

在您的情况下,为了使您的代码工作,只需创建一个File对象并命名您传入的文件(显然,这需要您知道本地文件的文件名,并将其硬编码到您的映射器代码中).该文件将位于当前工作目录中:

@Override
public void setup(Context context) throws IOException, InterruptedException {
  CsvReader csv = new CsvReader(new File("local-file.csv"));

  // work with csv file..

}