java - Hadoop:带有自定义输入格式的 NullPointerException

Question

我为 Hadoop 开发了一个自定义的 InputFormat(包括一个自定义的 InputSplit 和一个自定义的 RecordReader)，我正在经历一个罕见的NullPointerException。

这些类将用于查询第三方系统，该系统公开了用于记录检索的 REST API。于是，我在DBInputFormat中得到了启发。 ，这也是一个非 HDFS InputFormat。

我得到的错误如下:

Error: java.lang.NullPointerException at
org.apache.hadoop.mapred.MapTask$NewTrackingRecordReader.initialize(MapTask.java:524)
        at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:762)
        at org.apache.hadoop.mapred.MapTask.run(MapTask.java:339)
        at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:162)
        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:1491)
        at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:157)

我已经搜索了 MapTask 的代码(Hadoop 的 2.1.0 版本)我看到有问题的部分是 RecordReader 的初始化:

472 NewTrackingRecordReader(org.apache.hadoop.mapreduce.InputSplit split,
473       org.apache.hadoop.mapreduce.InputFormat<K, V> inputFormat,
474       TaskReporter reporter,
475       org.apache.hadoop.mapreduce.TaskAttemptContext taskContext)
476       throws InterruptedException, IOException {
...
491    this.real = inputFormat.createRecordReader(split, taskContext);
...
494 }
...
519 @Override
520 public void initialize(org.apache.hadoop.mapreduce.InputSplit split,
521       org.apache.hadoop.mapreduce.TaskAttemptContext context
522       ) throws IOException, InterruptedException {
523    long bytesInPrev = getInputBytes(fsStats);
524    real.initialize(split, context);
525    long bytesInCurr = getInputBytes(fsStats);
526    fileInputByteCounter.increment(bytesInCurr - bytesInPrev);
527 }

当然，我代码的相关部分:

# MyInputFormat.java

public static void setEnvironmnet(Job job, String host, String port, boolean ssl, String APIKey) {
    backend = new Backend(host, port, ssl, APIKey);
}

public static void addResId(Job job, String resId) {
    Configuration conf = job.getConfiguration();
    String inputs = conf.get(INPUT_RES_IDS, "");

    if (inputs.isEmpty()) {
        inputs += restId;
    } else {
        inputs += "," + resId;
    }

    conf.set(INPUT_RES_IDS, inputs);
}

@Override
public List<InputSplit> getSplits(JobContext job) {
    // resulting splits container
    List<InputSplit> splits = new ArrayList<InputSplit>();

    // get the Job configuration
    Configuration conf = job.getConfiguration();

    // get the inputs, i.e. the list of resource IDs
    String input = conf.get(INPUT_RES_IDS, "");
    String[] resIDs = StringUtils.split(input);

    // iterate on the resIDs
    for (String resID: resIDs) {
       splits.addAll(getSplitsResId(resID, job.getConfiguration()));
    }

    // return the splits
    return splits;
}

@Override
public RecordReader<LongWritable, Text> createRecordReader(InputSplit split, TaskAttemptContext context) {
    if (backend == null) {
        logger.info("Unable to create a MyRecordReader, it seems the environment was not properly set");
        return null;
    }

    // create a record reader
    return new MyRecordReader(backend, split, context);
}

# MyRecordReader.java

@Override
public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
    // get start, end and current positions
    MyInputSplit inputSplit = (MyInputSplit) this.split;
    start = inputSplit.getFirstRecordIndex();
    end = start + inputSplit.getLength();
    current = 0;

    // query the third-party system for the related resource, seeking to the start of the split
    records = backend.getRecords(inputSplit.getResId(), start, end);
}

# MapReduceTest.java

public static void main(String[] args) throws Exception {
    int res = ToolRunner.run(new Configuration(), new MapReduceTest(), args);
    System.exit(res);
}

@Override
public int run(String[] args) throws Exception {
    Configuration conf = this.getConf();
    Job job = Job.getInstance(conf, "MapReduce test");
    job.setJarByClass(MapReduceTest.class);
    job.setMapperClass(MyMap.class);
    job.setCombinerClass(MyReducer.class);
    job.setReducerClass(MyReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);
    job.setInputFormatClass(MyInputFormat.class);
    MyInputFormat.addInput(job, "ca73a799-9c71-4618-806e-7bd0ca1911f4");
    InputFormat.setEnvironmnet(job, "my.host.com", "443", true, "my_api_key");
    FileOutputFormat.setOutputPath(job, new Path(args[0]));
    return job.waitForCompletion(true) ? 0 : 1;
}

有什么问题吗？

顺便说一句，这是 RecordReader 必须使用的“好”InputSplit，它是给构造函数的那个​​，或者是在 initialize 中给定的那个> 方法？不管怎样，我已经尝试了这两个选项，结果错误是一样的:)

Answer 1

我读你的 strack trace real 的方式在第 524 行为 null。

但是不要相信我的话。在其中插入一个 assert 或 system.out.println 并自己检查 real 的值。

NullPointerException 几乎总是意味着您漏掉了您不希望为 null 的内容。一些图书馆和馆藏会把它扔给你，作为他们说“这不能为空”的方式。

Error: java.lang.NullPointerException at
org.apache.hadoop.mapred.MapTask$NewTrackingRecordReader.initialize(MapTask.java:524)

对我而言，它的意思是:在 org.apache.hadoop.mapred 包中，MapTask 类有一个内部类 NewTrackingRecordReader在第 524 行抛出 NullPointerException 的 initialize 方法。

524 real.initialize( blah, blah) // I actually stopped reading after the dot

this.real 在第 491 行设置。

491 this.real = inputFormat.createRecordReader(split, taskContext);

假设您没有遗漏任何范围更小的 real 来掩盖 this.real 那么我们需要查看 inputFormat.createRecordReader( split, taskContext); 如果这可以返回 null 那么它可能是罪魁祸首。

事实证明，当 backend 为 null 时，它将返回 null。

@Override
public RecordReader<LongWritable, Text> createRecordReader(
    InputSplit split, 
    TaskAttemptContext context) {

    if (backend == null) {
        logger.info("Unable to create a MyRecordReader, " + 
                    "it seems the environment was not properly set");
        return null;
    }

    // create a record reader
    return new MyRecordReader(backend, split, context);
}

看起来 setEnvironmnet 应该设置 backend

# MyInputFormat.java

public static void setEnvironmnet(
    Job job, 
    String host, 
    String port, 
    boolean ssl, 
    String APIKey) {

    backend = new Backend(host, port, ssl, APIKey);
}

backend 必须在 setEnvironment 之外的某处声明(否则您会遇到编译器错误)。

如果 backend 在构造时没有被设置为非空的东西并且 setEnvironmnet 在 createRecordReader 之前没有被调用那么你应该期望得到你得到的 NullPointerException。

更新:

正如您所注意到的，由于 setEnvironmnet() 是静态的，backend 也必须是静态的。这意味着您必须确保其他实例没有将其设置为空。