AbstractFileOutputOperator extends BaseOperator implements Operator.CheckpointListener, Operator.CheckpointNotificationListener { public void setFilterStreamProvider(FilterStreamProvider filterStreamProvider) { this.filterStreamProvider = filterStreamProvider; } 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 testGzipCompression() throws IOException { EvenOddHDFSExactlyOnceWriter writer = new EvenOddHDFSExactlyOnceWriter(); writer.setFilterStreamProvider(new FilterStreamCodec.GZipFilterStreamProvider()); File evenFile = new File(testMeta.getDir(), EVEN_FILE); File oddFile = new File(testMeta.getDir(), ODD_FILE); List<Long> evenOffsets = new ArrayList<Long>(); List<Long> oddOffsets = new ArrayList<Long>(); writeCompressedData(writer, evenFile, oddFile, evenOffsets, oddOffsets); checkCompressedFile(evenFile, evenOffsets, 0, 5, 1000, null, null); checkCompressedFile(oddFile, oddOffsets, 1, 5, 1000, null, null); }
@Test public void testSnappyStreamProvider() throws IOException { if (checkNativeSnappy()) { return; } EvenOddHDFSExactlyOnceWriter writer = new EvenOddHDFSExactlyOnceWriter(); writer.setFilterStreamProvider(new FilterStreamCodec.SnappyFilterStreamProvider()); File evenFile = new File(testMeta.getDir(), EVEN_FILE); File oddFile = new File(testMeta.getDir(), ODD_FILE); List<Long> evenOffsets = new ArrayList<Long>(); List<Long> oddOffsets = new ArrayList<Long>(); writeCompressedData(writer, evenFile, oddFile, evenOffsets, oddOffsets); checkSnappyFile(evenFile, evenOffsets, 0, 5, 1000); checkSnappyFile(oddFile, oddOffsets, 1, 5, 1000); }