FileBasedSink implements Serializable, HasDisplayData { @SuppressWarnings("unchecked") public DynamicDestinations<UserT, DestinationT, OutputT> getDynamicDestinations() { return (DynamicDestinations<UserT, DestinationT, OutputT>) dynamicDestinations; } @Experimental(Kind.FILESYSTEM)  FileBasedSink(
      ValueProvider<ResourceId> tempDirectoryProvider,
      DynamicDestinations<?, DestinationT, OutputT> dynamicDestinations); @Experimental(Kind.FILESYSTEM)  FileBasedSink(
      ValueProvider<ResourceId> tempDirectoryProvider,
      DynamicDestinations<?, DestinationT, OutputT> dynamicDestinations,
      WritableByteChannelFactory writableByteChannelFactory); @Experimental(Kind.FILESYSTEM)  FileBasedSink(
      ValueProvider<ResourceId> tempDirectoryProvider,
      DynamicDestinations<?, DestinationT, OutputT> dynamicDestinations,
      Compression compression); @Experimental(Kind.FILESYSTEM) static ResourceId convertToFileResourceIfPossible(String outputPrefix); @SuppressWarnings("unchecked") DynamicDestinations<UserT, DestinationT, OutputT> getDynamicDestinations(); @Experimental(Kind.FILESYSTEM) ValueProvider<ResourceId> getTempDirectoryProvider(); void validate(PipelineOptions options); abstract WriteOperation<DestinationT, OutputT> createWriteOperation(); @Override void populateDisplayData(DisplayData.Builder builder);  }
@Test public void testCopyToOutputFiles() throws Exception { SimpleSink.SimpleWriteOperation<Void> writeOp = buildWriteOperation(); List<String> inputFilenames = Arrays.asList("input-1", "input-2", "input-3"); List<String> inputContents = Arrays.asList("1", "2", "3"); List<String> expectedOutputFilenames = Arrays.asList("file-00-of-03.test", "file-01-of-03.test", "file-02-of-03.test"); List<KV<FileResult<Void>, ResourceId>> resultsToFinalFilenames = Lists.newArrayList(); List<ResourceId> expectedOutputPaths = Lists.newArrayList(); for (int i = 0; i < inputFilenames.size(); i++) { expectedOutputPaths.add( getBaseOutputDirectory() .resolve(expectedOutputFilenames.get(i), StandardResolveOptions.RESOLVE_FILE)); File inputTmpFile = tmpFolder.newFile(inputFilenames.get(i)); List<String> lines = Collections.singletonList(inputContents.get(i)); writeFile(lines, inputTmpFile); ResourceId finalFilename = writeOp .getSink() .getDynamicDestinations() .getFilenamePolicy(null) .unwindowedFilename(i, inputFilenames.size(), CompressionType.UNCOMPRESSED); resultsToFinalFilenames.add( KV.of( new FileResult<>( LocalResources.fromFile(inputTmpFile, false), UNKNOWN_SHARDNUM, GlobalWindow.INSTANCE, PaneInfo.ON_TIME_AND_ONLY_FIRING, null), finalFilename)); } writeOp.moveToOutputFiles(resultsToFinalFilenames); for (int i = 0; i < expectedOutputPaths.size(); i++) { assertFileContains( Collections.singletonList(inputContents.get(i)), expectedOutputPaths.get(i)); } }
@Test public void testGenerateOutputFilenames() { List<ResourceId> expected; List<ResourceId> actual; ResourceId root = getBaseOutputDirectory(); SimpleSink<Void> sink = SimpleSink.makeSimpleSink( root, "file", ".SSSSS.of.NNNNN", ".test", Compression.UNCOMPRESSED); FilenamePolicy policy = sink.getDynamicDestinations().getFilenamePolicy(null); expected = Arrays.asList( root.resolve("file.00000.of.00003.test", StandardResolveOptions.RESOLVE_FILE), root.resolve("file.00001.of.00003.test", StandardResolveOptions.RESOLVE_FILE), root.resolve("file.00002.of.00003.test", StandardResolveOptions.RESOLVE_FILE)); actual = generateDestinationFilenames(policy, 3); assertEquals(expected, actual); expected = Collections.singletonList( root.resolve("file.00000.of.00001.test", StandardResolveOptions.RESOLVE_FILE)); actual = generateDestinationFilenames(policy, 1); assertEquals(expected, actual); expected = new ArrayList<>(); actual = generateDestinationFilenames(policy, 0); assertEquals(expected, actual); }
@Test public void testGenerateOutputFilenamesWithoutExtension() { List<ResourceId> expected; List<ResourceId> actual; ResourceId root = getBaseOutputDirectory(); SimpleSink<Void> sink = SimpleSink.makeSimpleSink(root, "file", "-SSSSS-of-NNNNN", "", Compression.UNCOMPRESSED); FilenamePolicy policy = sink.getDynamicDestinations().getFilenamePolicy(null); expected = Arrays.asList( root.resolve("file-00000-of-00003", StandardResolveOptions.RESOLVE_FILE), root.resolve("file-00001-of-00003", StandardResolveOptions.RESOLVE_FILE), root.resolve("file-00002-of-00003", StandardResolveOptions.RESOLVE_FILE)); actual = generateDestinationFilenames(policy, 3); assertEquals(expected, actual); expected = Collections.singletonList( root.resolve("file-00000-of-00001", StandardResolveOptions.RESOLVE_FILE)); actual = generateDestinationFilenames(policy, 1); assertEquals(expected, actual); expected = new ArrayList<>(); actual = generateDestinationFilenames(policy, 0); assertEquals(expected, actual); }