@SuppressWarnings("deprecation") @Test public void output() throws Exception { final int count = 10000; LocalFileSystem fs = FileSystem.getLocal(conf); Path path = new Path(folder.newFile("testing").toURI()); try (ModelOutput<StringOption> out = format.createOutput(StringOption.class, fs, path, new Counter())) { StringOption value = new StringOption(); for (int i = 0; i < count; i++) { value.modify("Hello, world at " + i); out.write(value); } } try (SequenceFile.Reader reader = new SequenceFile.Reader(fs, path, conf)) { LongWritable k = new LongWritable(); Text v = new Text(); for (int i = 0; i < count; i++) { String answer = "Hello, world at " + i; assertThat(answer, reader.next(k, v), is(true)); assertThat(answer, k.get(), is(1L)); assertThat(answer, v.toString(), is(answer)); } assertThat("eof", reader.next(k), is(false)); } }
@Override public ModelOutput<T> createOutput( Class<? extends T> dataType, FileSystem fileSystem, Path path, Counter counter) throws IOException, InterruptedException { K keyBuffer = createKeyObject(); V valueBuffer = createValueObject(); CompressionCodec codec = getCompressionCodec(path); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Creating sequence file (path={0}, type={1}, codec={2})", path, dataType.getName(), codec)); } configure(codec); SequenceFile.Writer writer = SequenceFile.createWriter(getConf(), SequenceFile.Writer.file(fileSystem.makeQualified(path)), SequenceFile.Writer.keyClass(keyBuffer.getClass()), SequenceFile.Writer.valueClass(valueBuffer.getClass()), SequenceFile.Writer.compression(codec == null ? CompressionType.NONE : CompressionType.BLOCK, codec)); boolean succeed = false; try { ModelOutput<T> output = new ModelOutput<T>() { private long lastPosition = 0; @Override public void write(T model) throws IOException { copyFromModel(model, keyBuffer, valueBuffer); writer.append(keyBuffer, valueBuffer); long nextPosition = writer.getLength(); counter.add(nextPosition - lastPosition); lastPosition = nextPosition; } @Override public void close() throws IOException { writer.close(); } }; succeed = true; return output; } finally { if (succeed == false) { writer.close(); } } }
SequenceFileFormat extends HadoopFileFormat<T> { @Override public ModelOutput<T> createOutput( Class<? extends T> dataType, FileSystem fileSystem, Path path, Counter counter) throws IOException, InterruptedException { K keyBuffer = createKeyObject(); V valueBuffer = createValueObject(); CompressionCodec codec = getCompressionCodec(path); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Creating sequence file (path={0}, type={1}, codec={2})", path, dataType.getName(), codec)); } configure(codec); SequenceFile.Writer writer = SequenceFile.createWriter(getConf(), SequenceFile.Writer.file(fileSystem.makeQualified(path)), SequenceFile.Writer.keyClass(keyBuffer.getClass()), SequenceFile.Writer.valueClass(valueBuffer.getClass()), SequenceFile.Writer.compression(codec == null ? CompressionType.NONE : CompressionType.BLOCK, codec)); boolean succeed = false; try { ModelOutput<T> output = new ModelOutput<T>() { private long lastPosition = 0; @Override public void write(T model) throws IOException { copyFromModel(model, keyBuffer, valueBuffer); writer.append(keyBuffer, valueBuffer); long nextPosition = writer.getLength(); counter.add(nextPosition - lastPosition); lastPosition = nextPosition; } @Override public void close() throws IOException { writer.close(); } }; succeed = true; return output; } finally { if (succeed == false) { writer.close(); } } } }
SequenceFileFormat extends HadoopFileFormat<T> { @Override public ModelOutput<T> createOutput( Class<? extends T> dataType, FileSystem fileSystem, Path path, Counter counter) throws IOException, InterruptedException { K keyBuffer = createKeyObject(); V valueBuffer = createValueObject(); CompressionCodec codec = getCompressionCodec(path); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Creating sequence file (path={0}, type={1}, codec={2})", path, dataType.getName(), codec)); } configure(codec); SequenceFile.Writer writer = SequenceFile.createWriter(getConf(), SequenceFile.Writer.file(fileSystem.makeQualified(path)), SequenceFile.Writer.keyClass(keyBuffer.getClass()), SequenceFile.Writer.valueClass(valueBuffer.getClass()), SequenceFile.Writer.compression(codec == null ? CompressionType.NONE : CompressionType.BLOCK, codec)); boolean succeed = false; try { ModelOutput<T> output = new ModelOutput<T>() { private long lastPosition = 0; @Override public void write(T model) throws IOException { copyFromModel(model, keyBuffer, valueBuffer); writer.append(keyBuffer, valueBuffer); long nextPosition = writer.getLength(); counter.add(nextPosition - lastPosition); lastPosition = nextPosition; } @Override public void close() throws IOException { writer.close(); } }; succeed = true; return output; } finally { if (succeed == false) { writer.close(); } } }  }
SequenceFileFormat extends HadoopFileFormat<T> { @Override public ModelOutput<T> createOutput( Class<? extends T> dataType, FileSystem fileSystem, Path path, Counter counter) throws IOException, InterruptedException { K keyBuffer = createKeyObject(); V valueBuffer = createValueObject(); CompressionCodec codec = getCompressionCodec(path); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Creating sequence file (path={0}, type={1}, codec={2})", path, dataType.getName(), codec)); } configure(codec); SequenceFile.Writer writer = SequenceFile.createWriter(getConf(), SequenceFile.Writer.file(fileSystem.makeQualified(path)), SequenceFile.Writer.keyClass(keyBuffer.getClass()), SequenceFile.Writer.valueClass(valueBuffer.getClass()), SequenceFile.Writer.compression(codec == null ? CompressionType.NONE : CompressionType.BLOCK, codec)); boolean succeed = false; try { ModelOutput<T> output = new ModelOutput<T>() { private long lastPosition = 0; @Override public void write(T model) throws IOException { copyFromModel(model, keyBuffer, valueBuffer); writer.append(keyBuffer, valueBuffer); long nextPosition = writer.getLength(); counter.add(nextPosition - lastPosition); lastPosition = nextPosition; } @Override public void close() throws IOException { writer.close(); } }; succeed = true; return output; } finally { if (succeed == false) { writer.close(); } } }  @Override long getMinimumFragmentSize(); @Override ModelInput<T> createInput(
            Class<? extends T> dataType,
            FileSystem fileSystem,
            Path path,
            long offset,
            long fragmentSize,
            Counter counter); @Override ModelOutput<T> createOutput(
            Class<? extends T> dataType,
            FileSystem fileSystem,
            Path path,
            Counter counter); CompressionCodec getCompressionCodec(Path path); }
SequenceFileFormat extends HadoopFileFormat<T> { @Override public ModelOutput<T> createOutput( Class<? extends T> dataType, FileSystem fileSystem, Path path, Counter counter) throws IOException, InterruptedException { K keyBuffer = createKeyObject(); V valueBuffer = createValueObject(); CompressionCodec codec = getCompressionCodec(path); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Creating sequence file (path={0}, type={1}, codec={2})", path, dataType.getName(), codec)); } configure(codec); SequenceFile.Writer writer = SequenceFile.createWriter(getConf(), SequenceFile.Writer.file(fileSystem.makeQualified(path)), SequenceFile.Writer.keyClass(keyBuffer.getClass()), SequenceFile.Writer.valueClass(valueBuffer.getClass()), SequenceFile.Writer.compression(codec == null ? CompressionType.NONE : CompressionType.BLOCK, codec)); boolean succeed = false; try { ModelOutput<T> output = new ModelOutput<T>() { private long lastPosition = 0; @Override public void write(T model) throws IOException { copyFromModel(model, keyBuffer, valueBuffer); writer.append(keyBuffer, valueBuffer); long nextPosition = writer.getLength(); counter.add(nextPosition - lastPosition); lastPosition = nextPosition; } @Override public void close() throws IOException { writer.close(); } }; succeed = true; return output; } finally { if (succeed == false) { writer.close(); } } }  @Override long getMinimumFragmentSize(); @Override ModelInput<T> createInput(
            Class<? extends T> dataType,
            FileSystem fileSystem,
            Path path,
            long offset,
            long fragmentSize,
            Counter counter); @Override ModelOutput<T> createOutput(
            Class<? extends T> dataType,
            FileSystem fileSystem,
            Path path,
            Counter counter); CompressionCodec getCompressionCodec(Path path);  }