SequenceFile이 잘리는 이유는 무엇입니까?

Question

나는 Hadoop을 배우고 있으며,이 문제는 당황 스럽습니다. 기본적으로 디스크에 SequenceFile을 쓰고 다시 읽습니다. 그러나, 읽을 때마다 EOFException을 얻을 때마다. 자세히 살펴보면 시퀀스 파일을 작성할 때이 파일은 조기에 잘 리며 인덱스 962를 작성한 후에 항상 발생하며 파일의 크기는 항상 45056 바이트로 고정되어 있습니다.

MacBook Pro에서 Java 8 및 Hadoop 2.5.1을 사용하고 있습니다. 사실, 자바 7에서 다른 리눅스 머신에서 같은 코드를 시도했지만 똑같은 일이 일어납니다.

작가/리더가 제대로 닫히지 않을 수 있습니다. 이전 스타일의 try/catch 코드를 명시 적으로 writer.close() 함께 사용하여 시도했다 및 또한 최신 try-with-resource 방식을 사용하십시오. 둘 다 작동하지 않습니다.

도움이 될 것입니다.

public class SequenceFileDemo { 

private static final String[] DATA = { "One, two, buckle my shoe", 
    "Three, four, shut the door", 
    "Five, six, pick up sticks", 
    "Seven, eight, lay them straight", 
    "Nine, ten, a big fat hen" }; 

public static void main(String[] args) throws Exception { 
    String uri = "file:///Users/andy/Downloads/puzzling.seq"; 
    Configuration conf = new Configuration(); 
    FileSystem fs = FileSystem.get(URI.create(uri), conf); 

    Path path = new Path(uri);  
    IntWritable key = new IntWritable(); 
    Text value = new Text(); 

    //API change 
    try { 
     SequenceFile.Writer writer = SequenceFile.createWriter(conf, 
      stream(fs.create(path)), 
      keyClass(IntWritable.class), 
      valueClass(Text.class)); 

     for (int i = 0; i < 1024; i++) { 
      key.set(i); 
      value.clear(); 
      value.set(DATA[i % DATA.length]); 

      writer.append(key, value); 
      if ((i-1) %100 == 0) writer.hflush(); 
      System.out.printf("[%s]\t%s\t%s\n", writer.getLength(), key, value); 
     } 

     writer.close(); 

    } catch (Exception e) { 
     e.printStackTrace(); 
    } 


    try { 
     SequenceFile.Reader reader = new SequenceFile.Reader(conf, 
       SequenceFile.Reader.file(path)); 
     Class<?> keyClass = reader.getKeyClass(); 
     Class<?> valueClass = reader.getValueClass(); 

     boolean isWritableSerilization = false; 
     try { 
      keyClass.asSubclass(WritableComparable.class); 
      isWritableSerilization = true; 
     } catch (ClassCastException e) { 

     } 

     if (isWritableSerilization) { 
      WritableComparable<?> rKey = (WritableComparable<?>) ReflectionUtils.newInstance(keyClass, conf); 
      Writable rValue = (Writable) ReflectionUtils.newInstance(valueClass, conf); 
      while(reader.next(rKey, rValue)) { 
       System.out.printf("[%s] %d %s=%s\n",reader.syncSeen(), reader.getPosition(), rKey, rValue); 
      } 
     } else { 
      //make sure io.seraizliatons has the serialization in use when write the sequence file 
     } 

     reader.close(); 
    } catch(IOException e) { 
     e.printStackTrace(); 
    } 
} 

} 

Answer 1

실제로 오류를 발견했습니다. 이는 작성한 스트림을 절대로 Writer.stream(fs.create(path))에 닫지 않았기 때문입니다.

일부 이유 때문에 닫기는 방금 만든 스트림으로 전파되지 않습니다. 이것은 내가 생각하는 버그이지만, 나는 Jira에서 지금 그것을보기에는 너무 게으른 편이다.

문제를 해결하는 한 가지 방법은 대신 Writer.file(path)을 사용하는 것입니다.

분명히 스트림을 명시 적으로 닫을 수도 있습니다. 아래에서 수정 된 예를 찾으십시오.

Path path = new Path("file:///tmp/puzzling.seq"); 

    try (FSDataOutputStream stream = fs.create(path)) { 
     try (SequenceFile.Writer writer = SequenceFile.createWriter(conf, Writer.stream(stream), 
       Writer.keyClass(IntWritable.class), Writer.valueClass(NullWritable.class))) { 

      for (int i = 0; i < 1024; i++) { 
       writer.append(new IntWritable(i), NullWritable.get()); 
      } 
     } 
    } 

    try (SequenceFile.Reader reader = new SequenceFile.Reader(conf, Reader.file(path))) { 
     Class<?> keyClass = reader.getKeyClass(); 
     Class<?> valueClass = reader.getValueClass(); 

     WritableComparable<?> rKey = (WritableComparable<?>) ReflectionUtils.newInstance(keyClass, conf); 
     Writable rValue = (Writable) ReflectionUtils.newInstance(valueClass, conf); 
     while (reader.next(rKey, rValue)) { 
      System.out.printf("%s = %s\n", rKey, rValue); 
     } 

    } 

Answer 2

난 당신이 쓰기 루프 후 writer.close()을 누락 생각 :

다음

내가 사용하고있는 코드입니다. 독서를 시작하기 전에 최종 플러시를해야합니다.

Answer 3

Thomas에게 감사드립니다.

작가가 작성한 스트림이 "소유하고 있지 않은 경우"까지 계속됩니다. 작성자를 만들 때 Writer.file (경로) 옵션을 전달하면 작성자는 내부적으로 생성 된 기본 스트림을 "소유"하며 close()가 호출 될 때 작성자를 닫습니다. 그러나 Writer.stream (aStream)을 전달하면 작성자는 다른 사람이 해당 스트림에 대한 응답이며 close()가 호출 될 때 해당 스트림을 닫지 않는다고 가정합니다. 요컨대, 그것은 버그가 아니며 단지 충분히 잘 이해하지 못합니다. .