title: Beam on SparkRunner overwriting its own output
body: <p>I'm running a Beam pipeline on a SparkRunner, with Parquet file outputs (though the issue is there if I'm doing other IO outputs). The issue I'm having is that on output, the file copy is overwriting its own output. Here's the log output:</p>
  
  <pre><code>19/10/22 18:26:35 INFO FileBasedSink: Will copy temporary file FileResult{tempFilename=/home/hadoop/just1hour/.temp-beam-357d6916-5d8e-4519-a7a4-3852249011b5/77100cd1-04ae-441c-848f-e0d0067feeb8, shard=0, window=org.apache.beam.sdk.transforms.windowing.GlobalWindow@5316e95f,
   paneInfo=PaneInfo.NO_FIRING} to final location /home/hadoop/just1hour/output-00000-of-00001
  19/10/22 18:26:35 INFO FileBasedSink: Will copy temporary file FileResult{tempFilename=/home/hadoop/just1hour/.temp-beam-357d6916-5d8e-4519-a7a4-3852249011b5/f19819df-e006-431a-8ccd-6e67af692c3e, shard=0, window=org.apache.beam.sdk.transforms.windowing.GlobalWindow@5316e95f,
   paneInfo=PaneInfo.NO_FIRING} to final location /home/hadoop/just1hour/output-00000-of-00001
  19/10/22 18:26:35 INFO FileBasedSink: Will copy temporary file FileResult{tempFilename=/home/hadoop/just1hour/.temp-beam-357d6916-5d8e-4519-a7a4-3852249011b5/cb2abe0c-8cc2-4a94-ae54-97b67c5e7d20, shard=0, window=org.apache.beam.sdk.transforms.windowing.GlobalWindow@5316e95f,
   paneInfo=PaneInfo.NO_FIRING} to final location /home/hadoop/just1hour/output-00000-of-00002
  19/10/22 18:26:35 INFO FileBasedSink: Will copy temporary file FileResult{tempFilename=/home/hadoop/just1hour/.temp-beam-357d6916-5d8e-4519-a7a4-3852249011b5/611d194a-4e8f-44bc-8776-4bc2c55a8f34, shard=1, window=org.apache.beam.sdk.transforms.windowing.GlobalWindow@5316e95f,
   paneInfo=PaneInfo.NO_FIRING} to final location /home/hadoop/just1hour/output-00001-of-00002
  </code></pre>
  
  <p>As you can see the the first file is being overwritten. </p>
  
  <p>I was able to workaround this by manually specifying the number of shards to be equal to the number of input files, but I'm wondering if there's some other config that could explain or avert this behavior.</p>
  
  <p>Edit:</p>
  
  <p>This is a batch job and here's the code that generates the output:</p>
  
  <pre><code>p.apply(TextIO.read().from(input).withDelimiter("{".getBytes()))
                  .apply(Filter.by((String record) -&gt; !record.isEmpty()))
                  .apply(ParDo.of(new ParseNotificationJSON())).setCoder(AvroCoder.of(SCHEMA))
                  .apply("Write Parquet files",
                          FileIO.&lt;GenericRecord&gt;write().via(ParquetIO.sink(SCHEMA)).to(output));
  
          p.run().waitUntilFinish();
  </code></pre>