[HADOOP] Hadoop에서 사용자 정의 쓰기 가능 구현?

Hadoop에서 사용자 정의 쓰기 가능 구현?

Hadoop에서 사용자 정의 Writable 클래스를 정의했지만 Hadoop이 내 프로그램을 실행할 때 다음 오류 메시지를 표시합니다.

java.lang.RuntimeException: java.lang.NullPointerException
at org.apache.hadoop.util.ReflectionUtils.newInstance(ReflectionUtils.java:115)
at org.apache.hadoop.io.SortedMapWritable.readFields(SortedMapWritable.java:180)
at EquivalenceClsAggValue.readFields(EquivalenceClsAggValue.java:82)
at org.apache.hadoop.io.serializer.WritableSerialization$WritableDeserializer.deserialize(WritableSerialization.java:67)
at org.apache.hadoop.io.serializer.WritableSerialization$WritableDeserializer.deserialize(WritableSerialization.java:40)
at org.apache.hadoop.mapred.Task$ValuesIterator.readNextValue(Task.java:1282)
at org.apache.hadoop.mapred.Task$ValuesIterator.next(Task.java:1222)
at org.apache.hadoop.mapred.Task$CombineValuesIterator.next(Task.java:1301)
at Mondrian$Combine.reduce(Mondrian.java:119)
at Mondrian$Combine.reduce(Mondrian.java:1)
at org.apache.hadoop.mapred.Task$OldCombinerRunner.combine(Task.java:1442)
at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.sortAndSpill(MapTask.java:1436)
at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.flush(MapTask.java:1298)
at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:437)
at org.apache.hadoop.mapred.MapTask.run(MapTask.java:372)
at org.apache.hadoop.mapred.Child$4.run(Child.java:255)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:415)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1136)
at org.apache.hadoop.mapred.Child.main(Child.java:249)

원인 : java.lang.NullPointerException     java.util.concurrent.ConcurrentHashMap.hash (ConcurrentHashMap.java:332) ....

EquivalenceClsAggValue는 내가 정의한 Writable 클래스의 이름이며 이것은 내 클래스입니다.

public class EquivalenceClsAggValue implements WritableComparable<EquivalenceClsAggValue>{

public ArrayList<SortedMapWritable> aggValues;  
public EquivalenceClsAggValue(){        

    aggValues = new ArrayList<SortedMapWritable>();
}
@Override
public void readFields(DataInput arg0) throws IOException {

    int size = arg0.readInt();

    for (int i=0;i<size;i++){
        SortedMapWritable tmp = new SortedMapWritable();
        tmp.readFields(arg0);
        aggValues.add(tmp);
    }       
}

@Override
public void write(DataOutput arg0) throws IOException {

    //write the size first
    arg0.write(aggValues.size());

    //write each element
    for (SortedMapWritable s:aggValues){
        s.write(arg0);
    }

}

문제의 원인이 무엇인지 궁금합니다.

해결법

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

   1.write (DataOutput) 메서드에서 오류가 발생했습니다.

   write (DataOutput) 메서드에서 오류가 발생했습니다.

   @Override
   public void write(DataOutput arg0) throws IOException {
     //write the size first
     // arg0.write(aggValues.size()); // here you're writing an int as a byte
   
     // try this instead:
     arg0.writeInt(aggValues.size()); // actually write int as an int
   
     //..
   

   DataOutput.write (int) vs DataOutput.writeInt (int)에 대한 API 문서를 살펴보십시오.

   나는 또한 ReflectionUtils.newInstance ()를 사용하기 위해 readFields에 SortedMapWritable tmp 로컬 변수를 생성하도록 수정합니다.

   @Override
   public void readFields(DataInput arg0) throws IOException {
   
     int size = arg0.readInt();
   
     for (int i=0;i<size;i++){
       SortedMapWritable tmp = ReflectionUtils.newInstance(
           SortedMapWritable.class, getConf());
       tmp.readFields(arg0);
       aggValues.add(tmp);
     }       
   }
   

   이 기능이 작동하려면 다음과 같이 클래스 서명을 수정하여 Configurable을 확장해야합니다 (예 : 객체가 처음 생성 될 때 Hadoop이 Configuration 객체를 삽입 함).

   public class EquivalenceClsAggValue 
             extends Configured 
             implements WritableComparable<EquivalenceClsAggValue> {
   

from https://stackoverflow.com/questions/14249128/implementation-of-custom-writable-in-hadoop by cc-by-sa and MIT license