HadoopFileSystem extends FileSystem<HadoopResourceId> { @Override protected WritableByteChannel create(HadoopResourceId resourceId, CreateOptions createOptions) throws IOException { return Channels.newChannel( resourceId.toPath().getFileSystem(configuration).create(resourceId.toPath())); } HadoopFileSystem(String scheme, Configuration configuration);   }
@Test public void testCreateAndReadFile() throws Exception { byte[] bytes = "testData".getBytes(StandardCharsets.UTF_8); create("testFile", bytes); assertArrayEquals(bytes, read("testFile", 0)); }
@Test public void testCreateAndReadFileWithShift() throws Exception { byte[] bytes = "testData".getBytes(StandardCharsets.UTF_8); create("testFile", bytes); int bytesToSkip = 3; byte[] expected = Arrays.copyOfRange(bytes, bytesToSkip, bytes.length); byte[] actual = read("testFile", bytesToSkip); assertArrayEquals(expected, actual); }
@Test public void testCreateAndReadFileWithShiftToEnd() throws Exception { byte[] bytes = "testData".getBytes(StandardCharsets.UTF_8); create("testFile", bytes); int bytesToSkip = bytes.length; byte[] expected = Arrays.copyOfRange(bytes, bytesToSkip, bytes.length); assertArrayEquals(expected, read("testFile", bytesToSkip)); }
@Test @Ignore("TestPipeline needs a way to take in HadoopFileSystemOptions") public void testReadPipeline() throws Exception { create("testFileA", "testDataA".getBytes(StandardCharsets.UTF_8)); create("testFileB", "testDataB".getBytes(StandardCharsets.UTF_8)); create("testFileC", "testDataC".getBytes(StandardCharsets.UTF_8)); HadoopFileSystemOptions options = TestPipeline.testingPipelineOptions().as(HadoopFileSystemOptions.class); options.setHdfsConfiguration(ImmutableList.of(fileSystem.configuration)); FileSystems.setDefaultPipelineOptions(options); PCollection<String> pc = p.apply(TextIO.read().from(testPath("testFile*").toString())); PAssert.that(pc).containsInAnyOrder("testDataA", "testDataB", "testDataC"); p.run(); }