@Test public void testWrite() throws IOException { SimpleDataWriter writer = buildSimpleDataWriter(); int totalBytes = 3; for (String record : TestConstants.JSON_RECORDS) { byte[] toWrite = record.getBytes(); Assert.assertEquals(toWrite.length, record.length()); writer.write(toWrite); totalBytes += toWrite.length; } writer.close(); writer.commit(); Assert.assertEquals(writer.recordsWritten(), 3); Assert.assertEquals(writer.bytesWritten(), totalBytes); File outputFile = new File(writer.getOutputFilePath()); BufferedReader br = new BufferedReader(new FileReader(outputFile)); String line; int lineNumber = 0; while ((line = br.readLine()) != null) { Assert.assertEquals(line, TestConstants.JSON_RECORDS[lineNumber]); lineNumber++; } br.close(); Assert.assertEquals(lineNumber, 3); }
@Override public void write(byte[] record) throws IOException { Preconditions.checkNotNull(record); byte[] toWrite = record; if (this.recordDelimiter.isPresent()) { toWrite = Arrays.copyOf(record, record.length + 1); toWrite[toWrite.length - 1] = this.recordDelimiter.get(); } if (this.prependSize) { long recordSize = toWrite.length; ByteBuffer buf = ByteBuffer.allocate(Longs.BYTES); buf.putLong(recordSize); toWrite = ArrayUtils.addAll(buf.array(), toWrite); } this.stagingFileOutputStream.write(toWrite); this.bytesWritten += toWrite.length; this.recordsWritten++; }
SimpleDataWriter extends FsDataWriter<byte[]> { @Override public void write(byte[] record) throws IOException { Preconditions.checkNotNull(record); byte[] toWrite = record; if (this.recordDelimiter.isPresent()) { toWrite = Arrays.copyOf(record, record.length + 1); toWrite[toWrite.length - 1] = this.recordDelimiter.get(); } if (this.prependSize) { long recordSize = toWrite.length; ByteBuffer buf = ByteBuffer.allocate(Longs.BYTES); buf.putLong(recordSize); toWrite = ArrayUtils.addAll(buf.array(), toWrite); } this.stagingFileOutputStream.write(toWrite); this.bytesWritten += toWrite.length; this.recordsWritten++; } }
SimpleDataWriter extends FsDataWriter<byte[]> { @Override public void write(byte[] record) throws IOException { Preconditions.checkNotNull(record); byte[] toWrite = record; if (this.recordDelimiter.isPresent()) { toWrite = Arrays.copyOf(record, record.length + 1); toWrite[toWrite.length - 1] = this.recordDelimiter.get(); } if (this.prependSize) { long recordSize = toWrite.length; ByteBuffer buf = ByteBuffer.allocate(Longs.BYTES); buf.putLong(recordSize); toWrite = ArrayUtils.addAll(buf.array(), toWrite); } this.stagingFileOutputStream.write(toWrite); this.bytesWritten += toWrite.length; this.recordsWritten++; } SimpleDataWriter(SimpleDataWriterBuilder builder, State properties); }
SimpleDataWriter extends FsDataWriter<byte[]> { @Override public void write(byte[] record) throws IOException { Preconditions.checkNotNull(record); byte[] toWrite = record; if (this.recordDelimiter.isPresent()) { toWrite = Arrays.copyOf(record, record.length + 1); toWrite[toWrite.length - 1] = this.recordDelimiter.get(); } if (this.prependSize) { long recordSize = toWrite.length; ByteBuffer buf = ByteBuffer.allocate(Longs.BYTES); buf.putLong(recordSize); toWrite = ArrayUtils.addAll(buf.array(), toWrite); } this.stagingFileOutputStream.write(toWrite); this.bytesWritten += toWrite.length; this.recordsWritten++; } SimpleDataWriter(SimpleDataWriterBuilder builder, State properties); @Override void write(byte[] record); @Override long recordsWritten(); @Override long bytesWritten(); @Override boolean isSpeculativeAttemptSafe(); @Override void flush(); }
SimpleDataWriter extends FsDataWriter<byte[]> { @Override public void write(byte[] record) throws IOException { Preconditions.checkNotNull(record); byte[] toWrite = record; if (this.recordDelimiter.isPresent()) { toWrite = Arrays.copyOf(record, record.length + 1); toWrite[toWrite.length - 1] = this.recordDelimiter.get(); } if (this.prependSize) { long recordSize = toWrite.length; ByteBuffer buf = ByteBuffer.allocate(Longs.BYTES); buf.putLong(recordSize); toWrite = ArrayUtils.addAll(buf.array(), toWrite); } this.stagingFileOutputStream.write(toWrite); this.bytesWritten += toWrite.length; this.recordsWritten++; } SimpleDataWriter(SimpleDataWriterBuilder builder, State properties); @Override void write(byte[] record); @Override long recordsWritten(); @Override long bytesWritten(); @Override boolean isSpeculativeAttemptSafe(); @Override void flush();  }