java - Hadoop MultipleInputs 因 RuntimeException 而失败

Question

我的现有系统从特定文件夹读取所有文件，并在其上运行 MapReduce。代码如下:

    Path path = new Path(inputPath)
    if (!FileSystem.get(conf).exists(path)) {
      System.out.println("Path does not exist (skipping): " + path);
      return 1;
    }
    FileInputFormat.setInputPaths(conf, inputPath);

这运行没有任何问题。现在，最近的文件更改要求我指定将哪些文件用作输入。我将代码更改为:

for(String fileName:filePath.split(",")){
   MultipleInputs.addInputPath(conf, new Path(fileName), TextInputFormat.class, RawLogMapper.class);
   // MultipleInputs.addInputPath(conf, new Path(fileName), TextInputFormat.class);
}

其中 filePath 是需要处理的绝对文件路径的逗号分隔列表。我使用的是 mapred，而不是 mapreduce。

import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.lib.MultipleInputs;
import org.apache.hadoop.mapred.lib.MultipleTextOutputFormat;

更改代码后，我遇到了以下错误:

14/09/08 13:50:05 INFO mapred.JobClient: Task Id : attempt_201408201501_1196_m_000000_1, Status : FAILED
java.lang.RuntimeException: Error in configuring object
    at org.apache.hadoop.util.ReflectionUtils.setJobConf(ReflectionUtils.java:109)
    at org.apache.hadoop.util.ReflectionUtils.setConf(ReflectionUtils.java:75)
    at org.apache.hadoop.util.ReflectionUtils.newInstance(ReflectionUtils.java:133)
    at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:413)
    at org.apache.hadoop.mapred.MapTask.run(MapTask.java:332)
    at org.apache.hadoop.mapred.Child$4.run(Child.java:268)
    at java.security.AccessController.doPrivileged(Native Method)
    at javax.security.auth.Subject.doAs(Subject.java:396)
    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1438)
    at org.apache.hadoop.mapred.Child.main(Child.java:262)
Caused by: java.lang.reflect.InvocationTargetException
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.ja

不确定是不是因为我没有指定TextInputFormat，我把那部分加到addInputPath函数中，错误依旧。

编辑

发现问题。在映射器的下游进行了调用

String filename = conf.get("map.input.file");
pos = conf.get((new File(filename)).getName().split("-")[0]);

当我指定文件而不是给出文件夹时，文件名返回 null，因此是 NPE。我想知道为什么在我指定输入文件时使用 conf.get("map.input.file")。

https://issues.apache.org/jira/browse/MAPREDUCE-1743

这意味着，我需要在运行配置时知道文件的名称:

1. 不使用 conf.get("map.input.file")
2. 不使用 ((FileSplit) context.getInputSplit()).getPath().toString();，因为我使用的是 mapred 而不是 mapreduce。

public void configure(JobConf conf) {
  String filename = conf.get("map.input.file");
  merchant = conf.get((new File(filename)).getName().split("-")[0]);
  if (merchant == null) {
    merchant = "unknown_merchant";
  }
}

如果有任何意见可以解决此问题，我们将不胜感激。

谢谢，杰文

Answer 1

用过

FileInputFormat.addInputPath(conf, new Path(fileName));

一切看起来都很好。