FileBasedSource extends OffsetBasedSource<T> { @Override public final List<? extends FileBasedSource<T>> split( long desiredBundleSizeBytes, PipelineOptions options) throws Exception { String fileOrPattern = fileOrPatternSpec.get(); if (mode == Mode.FILEPATTERN) { long startTime = System.currentTimeMillis(); List<Metadata> expandedFiles = FileSystems.match(fileOrPattern, emptyMatchTreatment).metadata(); List<FileBasedSource<T>> splitResults = new ArrayList<>(expandedFiles.size()); for (Metadata metadata : expandedFiles) { FileBasedSource<T> split = createForSubrangeOfFile(metadata, 0, metadata.sizeBytes()); verify( split.getMode() == Mode.SINGLE_FILE_OR_SUBRANGE, "%s.createForSubrangeOfFile must return a source in mode %s", split, Mode.SINGLE_FILE_OR_SUBRANGE); splitResults.addAll(split.split(desiredBundleSizeBytes, options)); } LOG.info( "Splitting filepattern {} into bundles of size {} took {} ms " + "and produced {} files and {} bundles", fileOrPattern, desiredBundleSizeBytes, System.currentTimeMillis() - startTime, expandedFiles.size(), splitResults.size()); return splitResults; } else { if (isSplittable()) { @SuppressWarnings("unchecked") List<FileBasedSource<T>> splits = (List<FileBasedSource<T>>) super.split(desiredBundleSizeBytes, options); return splits; } else { LOG.debug( "The source for file {} is not split into sub-range based sources since " + "the file is not seekable", fileOrPattern); return ImmutableList.of(this); } } } protected  FileBasedSource(
      ValueProvider<String> fileOrPatternSpec,
      EmptyMatchTreatment emptyMatchTreatment,
      long minBundleSize); protected  FileBasedSource(ValueProvider<String> fileOrPatternSpec, long minBundleSize); protected  FileBasedSource(
      Metadata fileMetadata, long minBundleSize, long startOffset, long endOffset); final MatchResult.Metadata getSingleFileMetadata(); final String getFileOrPatternSpec(); final ValueProvider<String> getFileOrPatternSpecProvider(); final EmptyMatchTreatment getEmptyMatchTreatment(); final Mode getMode(); @Override final FileBasedSource<T> createSourceForSubrange(long start, long end); @Override final long getEstimatedSizeBytes(PipelineOptions options); @Override void populateDisplayData(DisplayData.Builder builder); @Override final List<? extends FileBasedSource<T>> split(
      long desiredBundleSizeBytes, PipelineOptions options); @Override final BoundedReader<T> createReader(PipelineOptions options); @Override String toString(); @Override void validate(); @Override final long getMaxEndOffset(PipelineOptions options);  }
@Test public void testSplittingFailsOnEmptyFileExpansion() throws Exception { PipelineOptions options = PipelineOptionsFactory.create(); String missingFilePath = tempFolder.newFolder().getAbsolutePath() + "/missing.txt"; TestFileBasedSource source = new TestFileBasedSource(missingFilePath, Long.MAX_VALUE, null); thrown.expect(FileNotFoundException.class); thrown.expectMessage(missingFilePath); source.split(1234, options); }
@Test public void testReadAllSplitsOfSingleFile() throws Exception { PipelineOptions options = PipelineOptionsFactory.create(); List<String> data = createStringDataset(3, 50); String fileName = "file"; File file = createFileWithData(fileName, data); TestFileBasedSource source = new TestFileBasedSource(file.getPath(), 16, null); List<? extends BoundedSource<String>> sources = source.split(32, null); assertTrue(sources.size() > 1); List<String> results = new ArrayList<>(); for (BoundedSource<String> split : sources) { results.addAll(readFromSource(split, options)); } assertThat(data, containsInAnyOrder(results.toArray())); }
@Test public void testReadAllSplitsOfFilePattern() throws Exception { PipelineOptions options = PipelineOptionsFactory.create(); List<String> data1 = createStringDataset(3, 50); File file1 = createFileWithData("file1", data1); List<String> data2 = createStringDataset(3, 50); createFileWithData("file2", data2); List<String> data3 = createStringDataset(3, 50); createFileWithData("file3", data3); List<String> data4 = createStringDataset(3, 50); createFileWithData("otherfile", data4); TestFileBasedSource source = new TestFileBasedSource(new File(file1.getParent(), "file*").getPath(), 64, null); List<? extends BoundedSource<String>> sources = source.split(512, null); assertTrue(sources.size() > 1); List<String> results = new ArrayList<>(); for (BoundedSource<String> split : sources) { results.addAll(readFromSource(split, options)); } List<String> expectedResults = new ArrayList<>(); expectedResults.addAll(data1); expectedResults.addAll(data2); expectedResults.addAll(data3); assertThat(expectedResults, containsInAnyOrder(results.toArray())); }