2

我正在学习Hadoop，这个问题让我困惑了一段时间。基本上我正在写一个SequenceFile磁盘然后读回来。但是，每次EOFException阅读时我都会得到一个。更深入的看会发现，在写序列文件时，会被提前截断，而且总是发生在写索引 962 之后，而且文件的大小总是固定的 45056 字节。

我在 MacBook Pro 上使用 Java 8 和 Hadoop 2.5.1。事实上，我在另一台 Java 7 下的 Linux 机器上尝试了相同的代码，但同样的事情发生了。

我可以排除作者/阅读器没有正确关闭。我尝试使用带有显式 writer.close() 的旧式 try/catch，如代码所示，并且还使用了较新的 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();
    }
}

}
4

3 回答 3

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);
        }

    }
于 2015-01-13T11:50:17.073 回答
1

I think you are missing writer.close() after write loop. That should gaurantee a final flush before you start reading.

于 2015-01-13T07:47:51.900 回答
0

感谢托马斯。

归结为作者是否创建了“拥有”非流。创建写入器时，如果我们传入选项Writer.file(path)，写入器“拥有”内部创建的底层流，并在调用 close() 时将其关闭。然而，如果我们传入Writer.stream(aStream)，则作者假定其他人是对该流的响应，并且在调用 close() 时不会关闭它。简而言之，这不是一个错误，只是我对它的理解不够好。.

于 2015-01-14T02:25:57.423 回答