hadoop - 如何在 Hadoop Mapreduce 作业中访问分布式缓存？

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。当我在伪分布式模式下运行它时效果很好，但是当我在集群上启 Action 业时似乎找不到该文件。我正在我的映射器的 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 中并将它们复制到任务节点，其中 as-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..