FileInputFormat extends RichInputFormat<OT, FileInputSplit> { protected FSDataInputStream decorateInputStream(FSDataInputStream inputStream, FileInputSplit fileSplit) throws Throwable { InflaterInputStreamFactory<?> inflaterInputStreamFactory = getInflaterInputStreamFactory(fileSplit.getPath()); if (inflaterInputStreamFactory != null) { return new InputStreamFSInputWrapper(inflaterInputStreamFactory.create(stream)); } return inputStream; } FileInputFormat(); protected  FileInputFormat(Path filePath); static void registerInflaterInputStreamFactory(String fileExtension, InflaterInputStreamFactory<?> factory); Path getFilePath(); void setFilePath(String filePath); void setFilePath(Path filePath); long getMinSplitSize(); void setMinSplitSize(long minSplitSize); int getNumSplits(); void setNumSplits(int numSplits); long getOpenTimeout(); void setOpenTimeout(long openTimeout); long getSplitStart(); long getSplitLength(); @Override void configure(Configuration parameters); @Override FileBaseStatistics getStatistics(BaseStatistics cachedStats); @Override LocatableInputSplitAssigner getInputSplitAssigner(FileInputSplit[] splits); @Override FileInputSplit[] createInputSplits(int minNumSplits); @Override void open(FileInputSplit fileSplit); @Override void close(); String toString(); static final String ENUMERATE_NESTED_FILES_FLAG; }
@Test public void testDecorateInputStream() throws IOException { final File tempFile = File.createTempFile("input-stream-decoration-test", "tmp"); tempFile.deleteOnExit(); final int blockSize = 8; final int numBlocks = 3; FileOutputStream fileOutputStream = new FileOutputStream(tempFile); for (int i = 0; i < blockSize * numBlocks; i++) { fileOutputStream.write(new byte[]{1}); } fileOutputStream.close(); final Configuration config = new Configuration(); final FileInputFormat<byte[]> inputFormat = new MyDecoratedInputFormat(); inputFormat.setFilePath(tempFile.toURI().toString()); inputFormat.configure(config); FileInputSplit[] inputSplits = inputFormat.createInputSplits(3); byte[] bytes = null; for (FileInputSplit inputSplit : inputSplits) { inputFormat.open(inputSplit); while (!inputFormat.reachedEnd()) { if ((bytes = inputFormat.nextRecord(bytes)) != null) { Assert.assertArrayEquals(new byte[]{(byte) 0xFE}, bytes); } } } }