AbstractFileOutputOperator extends BaseOperator implements Operator.CheckpointListener, Operator.CheckpointNotificationListener { protected void flush(FSDataOutputStream fsOutput) throws IOException { if (fs instanceof LocalFileSystem || fs instanceof RawLocalFileSystem) { fsOutput.flush(); } else { fsOutput.hflush(); } } AbstractFileOutputOperator(); @Override void setup(Context.OperatorContext context); @Override void teardown(); @Override void beginWindow(long windowId); @Override void endWindow(); void setFilePath(@Nonnull String dir); String getFilePath(); void setMaxLength(long maxLength); long getMaxLength(); int getRotationWindows(); void setRotationWindows(int rotationWindows); void setMaxOpenFiles(int maxOpenFiles); int getMaxOpenFiles(); short getFilePermission(); void setFilePermission(short filePermission); FilterStreamProvider getFilterStreamProvider(); void setFilterStreamProvider(FilterStreamProvider filterStreamProvider); @Override void beforeCheckpoint(long l); @Override void checkpointed(long l); @Override void committed(long l); boolean isAlwaysWriteToTmp(); void setAlwaysWriteToTmp(boolean alwaysWriteToTmp); Long getExpireStreamAfterAccessMillis(); void setExpireStreamAfterAccessMillis(Long millis); static final int DEFAULT_MAX_OPEN_FILES; final transient DefaultInputPort<INPUT> input; }
@Test public void testSnappyCompressionSimple() throws IOException { if (checkNativeSnappy()) { return; } File snappyFile = new File(testMeta.getDir(), "snappyTestFile.snappy"); BufferedOutputStream os = new BufferedOutputStream(new FileOutputStream(snappyFile)); Configuration conf = new Configuration(); CompressionCodec codec = (CompressionCodec)ReflectionUtils.newInstance(SnappyCodec.class, conf); FilterStreamCodec.SnappyFilterStream filterStream = new FilterStreamCodec.SnappyFilterStream( codec.createOutputStream(os)); int ONE_MB = 1024 * 1024; String testStr = "TestSnap-16bytes"; for (int i = 0; i < ONE_MB; i++) { filterStream.write(testStr.getBytes()); } filterStream.flush(); filterStream.close(); CompressionInputStream is = codec.createInputStream(new FileInputStream(snappyFile)); byte[] recovered = new byte[testStr.length()]; int bytesRead = is.read(recovered); is.close(); assertEquals(testStr, new String(recovered)); }