Merge pull request #240 from alxmrs/beam-no-fuse

Prevent operator fusion for every stage of the Beam executor.

pangeo_forge_recipes/executors/beam.py

@@ -62,7 +62,7 @@ def expand(self, pcoll):
 class BeamPipelineExecutor(PipelineExecutor[beam.PTransform]):
     @staticmethod
     def compile(pipeline: Pipeline) -> beam.PTransform:
-        pcoll = beam.Create([-1])
+        pcoll = "Start" >> beam.Create([-1])
         for step, stage in enumerate(pipeline.stages):
             if stage.mappable is not None:
                 pcoll |= stage.name >> _SingleArgumentStage(step, stage, pipeline.config)
@@ -71,6 +71,11 @@ def compile(pipeline: Pipeline) -> beam.PTransform:
                     _no_arg_stage, current=step, fun=stage.function, config=pipeline.config
                 )
 
+            # This prevents fusion:
+            #   https://cloud.google.com/dataflow/docs/guides/deploying-a-pipeline#preventing-fusion
+            # Avoiding fusion on Dataflow is necessary to ensure that stages execute serially.
+            pcoll |= f"Reshuffle_{step:03d}" >> beam.Reshuffle()
+
         return pcoll
 
     @staticmethod