[HADOOP] 하둡 : 사용자 정의 InputFormat와 NullPointerException이

하둡 : 사용자 정의 InputFormat와 NullPointerException이

나는 (사용자 정의 InputSplit 및 사용자 정의 RecordReader 포함) 하둡에 대한 사용자 정의 InputFormat을 개발했습니다 나는 드문 NullPointerException이 발생하고있다.

이러한 클래스는 기록 검색하기위한 REST API를 노출하는 타사 시스템을 쿼리에 사용되는 것입니다. 따라서, 나뿐만 아니라 비 HDFS InputFormat입니다 DBInputFormat에서 영감을 얻었다.

내가 오류는 다음과 같습니다

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에 대한 코드 (하둡의 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;
}

무엇이 잘못되었는지에 대한 어떤 아이디어?

BTW, 생성자 또는 초기화 방법에 주어진 일에 주어진 하나는 RecordReader 사용해야합니다 InputSplit은 "좋은"무엇입니까? 어쨌든 나는 두 옵션을 시도하고 그 결과 오류가 동일입니다 :)

해결법

1. ==============================

   1.나는 당신의 스택 추적 실제 읽기 방법은 라인 (524)에 null입니다.

   나는 당신의 스택 추적 실제 읽기 방법은 라인 (524)에 null입니다.

   그러나 내 말을하지 않습니다. 거기에 어설 또는에서 System.out.println 슬립과 실제 자신의 가치를 확인한다.

   NullPointerException가 거의 항상 당신이 null이 될 기대하지 않았다 뭔가를 점선을 의미합니다. 일부 라이브러리와 컬렉션 "이 null 일 수 없습니다"라는 자신의 방법으로 당신을 던질 것이다.

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

   나에게이 같은 읽고 다음 org.apache.hadoop.mapred 패키지의 MapTask 클래스는 라인 (524)에서 NullPointerException이 던져 초기화 방법 내부 클래스 NewTrackingRecordReader 있습니다.

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

   this.real는 라인 (491)에 설정되었다.

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

   당신은 우리가 inputFormat.createRecordReader (분할, taskContext) 볼 필요가 다음 this.real을 마스킹하는 더 밀접 범위 실수를 남겨하지 않은 가정하면, 이 널 (null)을 반환 할 수 있다면 그것은 범인 일 수 있습니다.

   백엔드가 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가 백엔드를 설정하도록되어 것 같습니다

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

   백엔드 setEnvironment 외부 어딘가에 선언해야합니다 (또는 당신은 컴파일러 오류가 발생 할 것).

   백엔드는 건설 및 setEnvironmnet에 따라 null 이외의 뭔가가 createRecordReader 전에 호출되지 않은 설정되어 있지 않은 경우에 당신은 당신이 가지고 정확히 NullPointerException이 얻을 것으로 예상한다.

   최신 정보:

   당신이 언급했듯이, setEnvironmnet ()가 정적 백엔드뿐만 아니라 정적이어야합니다. 이것은 당신이 다른 인스턴스가 null로 설정되지 않습니다 확인해야한다는 것을 의미합니다.

2. ==============================

   2.해결. 문제는 백엔드 변수는 자바 클래스에 속하며, 따라서 (예를 들어 널 (null)로) 그 변수를 변경하는 임의의 다른 목적은 동일한 클래스의 다른 모든 객체에 영향을 미치는, 즉, 정적으로 선언된다.

   해결. 문제는 백엔드 변수는 자바 클래스에 속하며, 따라서 (예를 들어 널 (null)로) 그 변수를 변경하는 임의의 다른 목적은 동일한 클래스의 다른 모든 객체에 영향을 미치는, 즉, 정적으로 선언된다.

   이제 setEnvironment는 호스트, 포트, SSL 사용 및 구성과 API 키 (이미 자원 ID와 함께했던 setResId보다 같은)를 추가; createRecordReader가 호출 될 때이 구성을 가지고 있으며 백엔드 개체가 만들어집니다.

   올바른 경로에 저를 넣어 CandiedOrange에 감사합니다!

from https://stackoverflow.com/questions/28213382/hadoop-nullpointerexception-with-custom-inputformat by cc-by-sa and MIT license