title: What is the ideal way to design this Apache Beam transform that outputs multiple files including binary outputs?
body: <p>I am trying to process PDF files in a Beam pipeline coming from an input bucket, and output the results, input, and intermediate file all to a separate output bucket.</p>
  
  <p><img src="https://i.imgur.com/YCaRNWq.png" alt="My pipeline"></p>
  
  <p>The filenames of all three outputs are derived from the final step, and there is a 1:1 mapping of input files to output filenames, so I don't want to have shard templates in the output filenames (my UniquePrefixFileNaming class is doing the same thing as TextIO.withoutSharding())</p>
  
  <p>Since the filenames are only known in the last step, I don't think I can set up tagged outputs and output files in each of the previous processing steps - I have to carry data all the way through the pipeline.</p>
  
  <p>What is the best way of achieving this?  Below is my attempt at the problem - the text outputs work okay but I don't have a solution for the PDF output (no binary output sink available, no binary data carried through).  Is FileIO.writeDynamic the best approach?</p>
  
  <pre class="lang-java prettyprint-override"><code>Pipeline p = Pipeline.create();
  
          PCollection&lt;MyProcessorTransformResult&gt; transformCollection = p.apply(FileIO.match().filepattern("Z:\\Inputs\\en_us\\**.pdf"))
                  .apply(FileIO.readMatches())
                  .apply(TikaIO.parseFiles())
                  .apply(ParDo.of(new MyProcessorTransform()));
  
          // Write output PDF
          transformCollection.apply(FileIO.match().filepattern())
          transformCollection.apply(FileIO.&lt;String, MyProcessorTransformResult&gt;writeDynamic()
                  .withTempDirectory("Z:\\Temp\\vbeam")
                  .by(input -&gt; input.data.getResourceKey())
                  .via(
                          Contextful.fn((SerializableFunction&lt;MyProcessorTransformResult, byte[]&gt;) input -&gt; new byte[] {})
                  )
                  .withNaming(d -&gt; new UniquePrefixFileNaming(d, ".pdf"))
                  .withNumShards(1)
                  .withDestinationCoder(ByteArrayCoder.of())
                  .to("Z:\\Outputs"));
  
          // Write output TXT
          transformCollection.apply(FileIO.&lt;String, MyProcessorTransformResult&gt;writeDynamic()
                  .withTempDirectory("Z:\\Temp\\vbeam")
                  .by(input -&gt; input.data.getResourceKey())
                  .via(
                          Contextful.fn((SerializableFunction&lt;MyProcessorTransformResult, String&gt;) input -&gt; input.originalContent),
                          TextIO.sink()
                  )
                  .withNaming(d -&gt; new UniquePrefixFileNaming(d, ".pdf.txt"))
                  .withNumShards(1)
                  .withDestinationCoder(StringUtf8Coder.of())
                  .to("Z:\\Outputs"));
  
          // Write output JSON
          transformCollection.apply(FileIO.&lt;String, MyProcessorTransformResult&gt;writeDynamic()
                  .withTempDirectory("Z:\\Temp\\vbeam")
                  .by(input -&gt; input.data.getResourceKey())
                  .via(
                          Contextful.fn((SerializableFunction&lt;MyProcessorTransformResult, String&gt;) input -&gt; SerializationHelpers.toJSON(input.data)),
                          TextIO.sink()
                  )
                  .withNaming(d -&gt; new UniquePrefixFileNaming(d, ".pdf.json"))
                  .withNumShards(1)
                  .withDestinationCoder(StringUtf8Coder.of())
                  .to("Z:\\Outputs"));
  
          p.run();
  </code></pre>